GH-518: Add clientIdPrefix to @KafkaListener

garyrussell · artembilan · commit a39780cae31a · 2017-12-20T13:03:53.000-05:00
Resolves #518 * Polishing - PR Comments
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListener.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListener.java
@@ -165,4 +165,13 @@
 	 */
 	boolean idIsGroup() default true;
 
+	/**
+	 * When provided, overrides the client id property in the consumer factory
+	 * configuration. A suffix ('-n') is added for each container instance to ensure
+	 * uniqueness when concurrency is used.
+	 * @return the client id prefix.
+	 * @since 2.1.1
+	 */
+	String clientIdPrefix() default "";
+
 }
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
@@ -383,6 +383,8 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 		endpoint.setTopicPartitions(resolveTopicPartitions(kafkaListener));
 		endpoint.setTopics(resolveTopics(kafkaListener));
 		endpoint.setTopicPattern(resolvePattern(kafkaListener));
+		endpoint.setClientIdPrefix(resolveExpressionAsString(kafkaListener.clientIdPrefix(),
+				"clientIdPrefix"));
 		String group = kafkaListener.containerGroup();
 		if (StringUtils.hasText(group)) {
 			Object resolvedGroup = resolveExpression(group);
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java
@@ -227,6 +227,7 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		endpoint.setupListenerContainer(instance, this.messageConverter);
 		initializeContainer(instance);
 		instance.getContainerProperties().setGroupId(endpoint.getGroupId());
+		instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());
 
 		return instance;
 	}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java
@@ -92,6 +92,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private KafkaTemplate<K, V> replyTemplate;
 
+	private String clientIdPrefix;
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -301,6 +303,21 @@ public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallb
 		this.recoveryCallback = recoveryCallback;
 	}
 
+	@Override
+	public String getClientIdPrefix() {
+		return this.clientIdPrefix;
+	}
+
+	/**
+	 * Set the client id prefix; overrides the client id in the consumer configuration
+	 * properties.
+	 * @param clientIdPrefix the prefix.
+	 * @since 2.1.1
+	 */
+	public void setClientIdPrefix(String clientIdPrefix) {
+		this.clientIdPrefix = clientIdPrefix;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		boolean topicsEmpty = getTopics().isEmpty();
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java
@@ -74,6 +74,15 @@ public interface KafkaListenerEndpoint {
 	 */
 	Pattern getTopicPattern();
 
+
+	/**
+	 * Return the client id prefix for the container; it will be suffixed by
+	 * '-n' to provide a unique id when concurrency is used.
+	 * @return the client id prefix.
+	 * @since 2.1.1
+	 */
+	String getClientIdPrefix();
+
 	/**
 	 * Setup the specified message listener container with the model
 	 * defined by this endpoint.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java
@@ -57,6 +57,23 @@ public interface ConsumerFactory<K, V> {
 	 */
 	Consumer<K, V> createConsumer(String groupId, String clientIdSuffix);
 
+	/**
+	 * Create a consumer with an explicit group id; in addition, the
+	 * client id suffix is appended to the clientIdPrefix which overrides the
+	 * {@code client.id} property, if present.
+	 * If a factory does not implement this method, {@link #createConsumer(String, String)}
+	 * is invoked, ignoring the prefix.
+	 * TODO: remove default in 2.2
+	 * @param groupId the group id.
+	 * @param clientIdPrefix the prefix.
+	 * @param clientIdSuffix the suffix.
+	 * @return the consumer.
+	 * @since 2.1.1
+	 */
+	default Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) {
+		return createConsumer(groupId, clientIdSuffix);
+	}
+
 	/**
 	 * Return true if consumers created by this factory use auto commit.
 	 * @return true if auto commit.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
@@ -25,6 +25,8 @@
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import org.springframework.util.StringUtils;
+
 /**
  * The {@link ConsumerFactory} implementation to produce a new {@link Consumer} instance
  * for provided {@link Map} {@code configs} and optional {@link Deserializer} {@code keyDeserializer},
@@ -95,13 +97,27 @@ public Consumer<K, V> createConsumer(String groupId, String clientIdSuffix) {
 		return createKafkaConsumer(groupId, clientIdSuffix);
 	}
 
+	@Override
+	public Consumer<K, V> createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) {
+		return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix);
+	}
+
 	protected KafkaConsumer<K, V> createKafkaConsumer() {
 		return createKafkaConsumer(this.configs);
 	}
 
 	protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientIdSuffix) {
-		boolean shouldModifyClientId = this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)
-				&& clientIdSuffix != null;
+		return createKafkaConsumer(groupId, null, clientIdSuffix);
+	}
+
+	protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientIdPrefix,
+			String clientIdSuffix) {
+		boolean overrideClientIdPrefix = StringUtils.hasText(clientIdPrefix);
+		if (clientIdSuffix == null) {
+			clientIdSuffix = "";
+		}
+		boolean shouldModifyClientId = (this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)
+				&& StringUtils.hasText(clientIdSuffix)) || overrideClientIdPrefix;
 		if (groupId == null && !shouldModifyClientId) {
 			return createKafkaConsumer();
 		}
@@ -112,7 +128,8 @@ protected KafkaConsumer<K, V> createKafkaConsumer(String groupId, String clientI
 			}
 			if (shouldModifyClientId) {
 				modifiedConfigs.put(ConsumerConfig.CLIENT_ID_CONFIG,
-					modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG) + clientIdSuffix);
+					(overrideClientIdPrefix ? clientIdPrefix
+							: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
 			}
 			return createKafkaConsumer(modifiedConfigs);
 		}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
@@ -387,8 +387,11 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 		ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
 			Assert.state(!this.isAnyManualAck || !this.autoCommit,
 					"Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode());
-			final Consumer<K, V> consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
-					this.consumerGroupId, KafkaMessageListenerContainer.this.clientIdSuffix);
+			final Consumer<K, V> consumer =
+					KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
+							this.consumerGroupId,
+							this.containerProperties.getClientId(),
+							KafkaMessageListenerContainer.this.clientIdSuffix);
 			this.consumer = consumer;
 
 			ConsumerRebalanceListener rebalanceListener = createRebalanceListener(consumer);
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java
@@ -157,6 +157,8 @@ public class ContainerProperties {
 
 	private float noPollThreshold = DEFAULT_NO_POLL_THRESHOLD;
 
+	private String clientId = "";
+
 	public ContainerProperties(String... topics) {
 		Assert.notEmpty(topics, "An array of topicPartitions must be provided");
 		this.topics = Arrays.asList(topics).toArray(new String[topics.length]);
@@ -460,4 +462,25 @@ public void setNoPollThreshold(float noPollThreshold) {
 		this.noPollThreshold = noPollThreshold;
 	}
 
+	/**
+	 * Return the client id.
+	 * @return the client id.
+	 * @since 2.1.1
+	 * @see #setClientId(String)
+	 */
+	public String getClientId() {
+		return this.clientId;
+	}
+
+	/**
+	 * Set the client id; overrides the consumer factory client.id property.
+	 * When used in a concurrent container, will be suffixed with '-n' to
+	 * provide a unique value for each consumer.
+	 * @param clientId the client id.
+	 * @since 2.1.1
+	 */
+	public void setClientId(String clientId) {
+		this.clientId = clientId;
+	}
+
 }
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
@@ -218,6 +218,8 @@ public void testSimple() throws Exception {
 		List<?> containers = KafkaTestUtils.getPropertyValue(manualContainer, "containers", List.class);
 		assertThat(KafkaTestUtils.getPropertyValue(containers.get(0), "listenerConsumer.consumerGroupId"))
 				.isEqualTo("qux");
+		assertThat(KafkaTestUtils.getPropertyValue(containers.get(0), "listenerConsumer.consumer.clientId"))
+				.isEqualTo("clientIdViaProps3-0");
 
 		template.send("annotated5", 0, 0, "foo");
 		template.send("annotated5", 1, 0, "bar");
@@ -237,6 +239,8 @@ public void testSimple() throws Exception {
 		assertThat(offset.isRelativeToCurrent()).isTrue();
 		assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.coordinator.groupId"))
 				.isEqualTo("fiz");
+		assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.clientId"))
+				.isEqualTo("clientIdViaAnnotation-0");
 
 		template.send("annotated11", 0, "foo");
 		template.flush();
@@ -251,6 +255,13 @@ public void testSimple() throws Exception {
 				.getPropertyValue(rebalanceConcurrentContainer, "containers", List.class).get(0);
 		assertThat(KafkaTestUtils.getPropertyValue(rebalanceContainer, "listenerConsumer.consumer.coordinator.groupId"))
 				.isNotEqualTo("rebalanceListener");
+		String clientId = KafkaTestUtils.getPropertyValue(rebalanceContainer, "listenerConsumer.consumer.clientId",
+			String.class);
+		assertThat(
+				clientId)
+				.startsWith("consumer-");
+		assertThat(clientId.indexOf('-')).isEqualTo(clientId.lastIndexOf('-'));
+
 	}
 
 	@Test
@@ -673,7 +684,8 @@ public KafkaListenerContainerFactory<?> batchSpyFactory() {
 			ConsumerFactory spiedCf = mock(ConsumerFactory.class);
 			willAnswer(i -> {
 				Consumer<Integer, String> spy =
-						spy(consumerFactory().createConsumer(i.getArgument(0), i.getArgument(1)));
+						spy(consumerFactory().createConsumer(i.getArgument(0), i.getArgument(1),
+								i.getArgument(2)));
 				willAnswer(invocation -> {
 
 					try {
@@ -685,7 +697,7 @@ public KafkaListenerContainerFactory<?> batchSpyFactory() {
 
 				}).given(spy).commitSync(anyMap());
 				return spy;
-			}).given(spiedCf).createConsumer(anyString(), anyString());
+			}).given(spiedCf).createConsumer(anyString(), anyString(), anyString());
 			factory.setConsumerFactory(spiedCf);
 			factory.setBatchListener(true);
 			factory.setRecordFilterStrategy(recordFilter());
@@ -698,7 +710,18 @@ public KafkaListenerContainerFactory<?> batchSpyFactory() {
 		public KafkaListenerContainerFactory<?> batchManualFactory() {
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
-			factory.setConsumerFactory(manualConsumerFactory());
+			factory.setConsumerFactory(manualConsumerFactory("clientIdViaProps1"));
+			ContainerProperties props = factory.getContainerProperties();
+			props.setAckMode(AckMode.MANUAL_IMMEDIATE);
+			factory.setBatchListener(true);
+			return factory;
+		}
+
+		@Bean
+		public KafkaListenerContainerFactory<?> batchManualFactory2() {
+			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(manualConsumerFactory("clientIdViaProps2"));
 			ContainerProperties props = factory.getContainerProperties();
 			props.setAckMode(AckMode.MANUAL_IMMEDIATE);
 			factory.setBatchListener(true);
@@ -710,7 +733,7 @@ public KafkaListenerContainerFactory<?> batchManualFactory() {
 				kafkaManualAckListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
-			factory.setConsumerFactory(manualConsumerFactory());
+			factory.setConsumerFactory(manualConsumerFactory("clientIdViaProps3"));
 			ContainerProperties props = factory.getContainerProperties();
 			props.setAckMode(AckMode.MANUAL_IMMEDIATE);
 			props.setIdleEventInterval(100L);
@@ -751,10 +774,10 @@ public DefaultKafkaConsumerFactory<Integer, String> consumerFactory() {
 			return new DefaultKafkaConsumerFactory<>(consumerConfigs());
 		}
 
-		@Bean
-		public ConsumerFactory<Integer, String> manualConsumerFactory() {
+		private ConsumerFactory<Integer, String> manualConsumerFactory(String clientId) {
 			Map<String, Object> configs = consumerConfigs();
 			configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+			configs.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
 			return new DefaultKafkaConsumerFactory<>(configs);
 		}
 
@@ -1090,7 +1113,7 @@ public void eventHandler(ListenerContainerIdleEvent event) {
 				@TopicPartition(topic = "annotated6", partitions = "0",
 						partitionOffsets = @PartitionOffset(partition = "${xxx:1}", initialOffset = "${yyy:0}",
 								relativeToCurrent = "${zzz:true}"))
-		})
+		}, clientIdPrefix = "${foo.xxx:clientIdViaAnnotation}")
 		public void listen5(ConsumerRecord<?, ?> record) {
 			this.record = record;
 			this.latch5.countDown();
@@ -1183,7 +1206,7 @@ public void listen14(List<Message<?>> list) {
 			this.latch14.countDown();
 		}
 
-		@KafkaListener(id = "list6", topics = "annotated19", containerFactory = "batchManualFactory")
+		@KafkaListener(id = "list6", topics = "annotated19", containerFactory = "batchManualFactory2")
 		public void listen15(List<Message<?>> list, Acknowledgment ack) {
 			this.payload = list;
 			this.ack = ack;
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
@@ -393,7 +393,7 @@ public void testConcurrencyWithPartitions() {
 		};
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
 		Consumer<Integer, String> consumer = mock(Consumer.class);
-		given(cf.createConsumer(isNull(), anyString())).willReturn(consumer);
+		given(cf.createConsumer(isNull(), anyString(), anyString())).willReturn(consumer);
 		given(consumer.poll(anyLong()))
 			.willAnswer(new Answer<ConsumerRecords<Integer, String>>() {
 
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java
@@ -122,7 +122,7 @@ public void foo(List<String> in) {
 		public ConsumerFactory consumerFactory() {
 			ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
 			final Consumer consumer = consumer();
-			given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer);
+			given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0")).willReturn(consumer);
 			return consumerFactory;
 		}
 
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java
@@ -132,7 +132,7 @@ public void foo(String in) {
 		public ConsumerFactory consumerFactory() {
 			ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
 			final Consumer consumer = consumer();
-			given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer);
+			given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0")).willReturn(consumer);
 			return consumerFactory;
 		}
 
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java
@@ -142,7 +142,7 @@ public void foo(String in) {
 		public ConsumerFactory consumerFactory() {
 			ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
 			final Consumer consumer = consumer();
-			given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer);
+			given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0")).willReturn(consumer);
 			return consumerFactory;
 		}
 
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
@@ -19,6 +19,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
@@ -490,7 +491,7 @@ public void testRecordAck() throws Exception {
 	public void testRecordAckMock() throws Exception {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
 		Consumer<Integer, String> consumer = mock(Consumer.class);
-		given(cf.createConsumer(isNull(), isNull())).willReturn(consumer);
+		given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer);
 		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
 		records.put(new TopicPartition("foo", 0), Arrays.asList(
 				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
@@ -527,6 +528,7 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		).given(consumer).commitSync(any(Map.class));
 
 		containerProps.setMessageListener(messageListener);
+		containerProps.setClientId("clientId");
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.start();
@@ -546,7 +548,7 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 	public void testNonResponsiveConsumerEvent() throws Exception {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
 		Consumer<Integer, String> consumer = mock(Consumer.class);
-		given(cf.createConsumer(isNull(), isNull())).willReturn(consumer);
+		given(cf.createConsumer(isNull(), eq(""), isNull())).willReturn(consumer);
 		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
 		records.put(new TopicPartition("foo", 0), Arrays.asList(
 				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
@@ -1060,7 +1062,8 @@ public void testDefinedPartitions() throws Exception {
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
 
 			@Override
-			public Consumer<Integer, String> createConsumer(String groupId, String clientIdPrefix) {
+			public Consumer<Integer, String> createConsumer(String groupId, String clientIdPrefix,
+					String clientIdSufffix) {
 				return new KafkaConsumer<Integer, String>(props) {
 
 					@Override
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java
@@ -138,7 +138,7 @@ public void foo(List<String> in) {
 		public ConsumerFactory consumerFactory() {
 			ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
 			final Consumer consumer = consumer();
-			given(consumerFactory.createConsumer(CONTAINER_ID, "-0")).willReturn(consumer);
+			given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0")).willReturn(consumer);
 			return consumerFactory;
 		}
 
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java
diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc

Original file line number	Diff line number	Diff line change
`@@ -227,6 +227,7 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {`
`227`	`227`	`endpoint.setupListenerContainer(instance, this.messageConverter);`
`228`	`228`	`initializeContainer(instance);`
`229`	`229`	`instance.getContainerProperties().setGroupId(endpoint.getGroupId());`
	`230`	`+ instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());`
`230`	`231`
`231`	`232`	`return instance;`
`232`	`233`	`}`