SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Kafka Binder uses a fixed-size daemon thread pool

Browse Source 
Fix #500

- Initialize a fixed-size ExecutorService of the size of the concurrency setting of the binder
- Clean up the ExecutorService on unbind
- Remove unused `spyOn` support
This commit is contained in:
Marius Bogoevici
2016-04-26 16:06:47 -04:00
committed by Mark Fisher
parent 6227e24e5d
commit 0bd7899a77
2 changed files with 43 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -27,6 +27,9 @@ import java.util.LinkedList;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import kafka.admin.AdminUtils;
@@ -106,6 +109,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ex
public static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer();
public static final DaemonThreadFactory DAEMON_THREAD_FACTORY = new DaemonThreadFactory();
private RetryOperations retryOperations;
private final Map<String, Collection<Partition>> topicsInUse = new HashMap<>();
@@ -460,9 +465,27 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ex
final FixedSubscriberChannel bridge = new FixedSubscriberChannel(rh);
bridge.setBeanName("bridge." + name);
final KafkaMessageListenerContainer messageListenerContainer =
createMessageListenerContainer(properties, group, null, listenedPartitions, referencePoint);
Assert.isTrue(!CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided");
final KafkaMessageListenerContainer messageListenerContainer = new KafkaMessageListenerContainer(connectionFactory,
listenedPartitions.toArray(new Partition[listenedPartitions.size()]));
if (logger.isDebugEnabled()) {
logger.debug("Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions));
}
OffsetManager offsetManager = createOffsetManager(group, referencePoint);
if (properties.getExtension().isResetOffsets()) {
offsetManager.resetOffsets(listenedPartitions);
}
messageListenerContainer.setOffsetManager(offsetManager);
messageListenerContainer.setQueueSize(queueSize);
messageListenerContainer.setMaxFetch(fetchSize);
int concurrency = Math.min(properties.getConcurrency(), listenedPartitions.size());
messageListenerContainer.setConcurrency(concurrency);
final ExecutorService dispatcherTaskExecutor = Executors.newFixedThreadPool(concurrency, DAEMON_THREAD_FACTORY);
messageListenerContainer.setDispatcherTaskExecutor(dispatcherTaskExecutor);
final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter(messageListenerContainer);
kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());
@@ -495,40 +518,17 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ex
String groupedName = groupedName(name, group);
edc.setBeanName("inbound." + groupedName);
DefaultBinding<MessageChannel> consumerBinding = new DefaultBinding<>(name, group, moduleInputChannel, edc);
DefaultBinding<MessageChannel> consumerBinding = new DefaultBinding<MessageChannel>(name, group,
moduleInputChannel, edc) {
@Override
protected void afterUnbind() {
dispatcherTaskExecutor.shutdown();
}
};
edc.start();
return consumerBinding;
}
KafkaMessageListenerContainer createMessageListenerContainer(
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties,
String group, String topic, Collection<Partition> listenedPartitions,
long referencePoint) {
Assert.isTrue(StringUtils.hasText(topic) ^ !CollectionUtils.isEmpty(listenedPartitions),
"Exactly one of topic or a list of listened partitions must be provided");
KafkaMessageListenerContainer messageListenerContainer;
if (topic != null) {
messageListenerContainer = new KafkaMessageListenerContainer(connectionFactory, topic);
}
else {
messageListenerContainer = new KafkaMessageListenerContainer(connectionFactory,
listenedPartitions.toArray(new Partition[listenedPartitions.size()]));
}
if (logger.isDebugEnabled()) {
logger.debug("Listening to topic " + topic);
}
// if we have fewer target partitions than target concurrency, adjust accordingly
messageListenerContainer.setConcurrency(Math.min(consumerProperties.getConcurrency(), listenedPartitions.size()));
OffsetManager offsetManager = createOffsetManager(group, referencePoint);
if (consumerProperties.getExtension().isResetOffsets()) {
offsetManager.resetOffsets(listenedPartitions);
}
messageListenerContainer.setOffsetManager(offsetManager);
messageListenerContainer.setQueueSize(queueSize);
messageListenerContainer.setMaxFetch(fetchSize);
return messageListenerContainer;
}
private OffsetManager createOffsetManager(String group, long referencePoint) {
try {
@@ -564,6 +564,16 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel, Ex
}
}
private static class DaemonThreadFactory implements ThreadFactory {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "kafka-binder-");
thread.setDaemon(true);
return thread;
}
}
private class ReceivingHandler extends AbstractReplyProducingMessageHandler {
private ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties;

35
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -28,11 +28,7 @@ import static org.junit.Assert.assertTrue;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import kafka.api.OffsetRequest;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
@@ -47,10 +43,7 @@ import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.integration.kafka.listener.MessageListener;
import org.springframework.integration.kafka.support.ProducerConfiguration;
import org.springframework.integration.kafka.support.ProducerMetadata;
import org.springframework.integration.kafka.support.ZookeeperConnect;
@@ -119,33 +112,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
@Override
public Spy spyOn(final String name) {
KafkaMessageChannelBinder.validateTopicName(name);
KafkaTestBinder binderWrapper = getBinder();
// Rewind offset, as tests will have typically already sent the messages we're trying to consume
KafkaMessageListenerContainer messageListenerContainer = binderWrapper.getCoreBinder().createMessageListenerContainer(
createConsumerProperties(), UUID.randomUUID().toString(), name, null, OffsetRequest.EarliestTime());
final BlockingQueue<KafkaMessage> messages = new ArrayBlockingQueue<KafkaMessage>(10);
messageListenerContainer.setMessageListener(new MessageListener() {
@Override
public void onMessage(KafkaMessage message) {
messages.offer(message);
}
});
return new Spy() {
@Override
public Object receive(boolean expectNull) throws Exception {
return messages.poll(expectNull ? 50 : 5000, TimeUnit.MILLISECONDS);
}
};
throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests");
}
@Test(expected = IllegalArgumentException.class)