diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index 42a084b07..3b2a25282 100644
--- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -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;
diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
index 0f2fada78..776ea7b91 100644
--- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
+++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
@@ -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)