diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index b153293bc..7055f120e 100644
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -623,6 +623,7 @@ public class KafkaMessageChannelBinder extends
 	@Override
 	protected PolledConsumerResources createPolledConsumerResources(String name, String group,
 			ConsumerDestination destination, ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
+
 		boolean anonymous = !StringUtils.hasText(group);
 		Assert.isTrue(!anonymous || !consumerProperties.getExtension().isEnableDlq(),
 				"DLQ support is not available for anonymous subscriptions");
@@ -638,6 +639,11 @@ public class KafkaMessageChannelBinder extends
 		KafkaMessageSource<?, ?> source = new KafkaMessageSource<>(consumerFactory, topics);
 		source.setMessageConverter(getMessageConverter(consumerProperties));
 		source.setRawMessageHeader(consumerProperties.getExtension().isEnableDlq());
+		String clientId = name;
+		if (consumerProperties.getExtension().getConfiguration().containsKey(ConsumerConfig.CLIENT_ID_CONFIG)) {
+			clientId = consumerProperties.getExtension().getConfiguration().get(ConsumerConfig.CLIENT_ID_CONFIG);
+		}
+		source.setClientId(clientId);
 
 		if (!consumerProperties.isMultiplex()) {
 			// I copied this from the regular consumer - it looks bogus to me - includes all partitions
diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
index ee86890a5..be7afc4a2 100644
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -35,6 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
@@ -2576,7 +2577,14 @@ public class KafkaBinderTests extends
 			Thread.sleep(100);
 		}
 		assertThat(polled).isTrue();
+		// Bind a second pollable consumer GH-521
+		consumerProps.getExtension().getConfiguration().put(ConsumerConfig.CLIENT_ID_CONFIG, "pollable2");
+		PollableSource<MessageHandler> second = new DefaultPollableMessageSource(this.messageConverter);
+		Binding<PollableSource<MessageHandler>> binding2 = binder.bindPollableConsumer("pollable2",
+				"group-polledConsumer2", second, consumerProps);
+		second.poll(m -> { });
 		binding.unbind();
+		binding2.unbind();
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })