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 index 44130c3e..2360f72b 100644 --- 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 @@ public @interface KafkaListener { */ 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 index 64fe5645..e1201092 100644 --- 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 @@ public class KafkaListenerAnnotationBeanPostProcessor 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 index f1a5ebcc..7d341652 100644 --- 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 abstract class AbstractKafkaListenerContainerFactory private KafkaTemplate replyTemplate; + private String clientIdPrefix; + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { this.beanFactory = beanFactory; @@ -301,6 +303,21 @@ public abstract class AbstractKafkaListenerEndpoint 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 index 77753e89..91cb4044 100644 --- 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 index 6c59ef69..b88cb8c0 100644 --- 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 { */ Consumer 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 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 index 5f7305c6..1664e58f 100644 --- 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.ConsumerConfig; 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 class DefaultKafkaConsumerFactory implements ConsumerFactory return createKafkaConsumer(groupId, clientIdSuffix); } + @Override + public Consumer createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) { + return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix); + } + protected KafkaConsumer createKafkaConsumer() { return createKafkaConsumer(this.configs); } protected KafkaConsumer createKafkaConsumer(String groupId, String clientIdSuffix) { - boolean shouldModifyClientId = this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG) - && clientIdSuffix != null; + return createKafkaConsumer(groupId, null, clientIdSuffix); + } + + protected KafkaConsumer 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 @@ public class DefaultKafkaConsumerFactory implements ConsumerFactory } 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 index 2dbdd135..df197ce4 100644 --- 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 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener 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 consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer( - this.consumerGroupId, KafkaMessageListenerContainer.this.clientIdSuffix); + final Consumer 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 index a7b0d82a..93ef26e7 100644 --- 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 class ContainerProperties { 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 index 5b999eb8..eca83568 100644 --- 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 class EnableKafkaIntegrationTests { 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 class EnableKafkaIntegrationTests { 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 class EnableKafkaIntegrationTests { .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 class EnableKafkaIntegrationTests { ConsumerFactory spiedCf = mock(ConsumerFactory.class); willAnswer(i -> { Consumer 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 class EnableKafkaIntegrationTests { }).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 class EnableKafkaIntegrationTests { public KafkaListenerContainerFactory batchManualFactory() { ConcurrentKafkaListenerContainerFactory 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 factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(manualConsumerFactory("clientIdViaProps2")); ContainerProperties props = factory.getContainerProperties(); props.setAckMode(AckMode.MANUAL_IMMEDIATE); factory.setBatchListener(true); @@ -710,7 +733,7 @@ public class EnableKafkaIntegrationTests { kafkaManualAckListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory 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 class EnableKafkaIntegrationTests { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } - @Bean - public ConsumerFactory manualConsumerFactory() { + private ConsumerFactory manualConsumerFactory(String clientId) { Map 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 class EnableKafkaIntegrationTests { @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 class EnableKafkaIntegrationTests { this.latch14.countDown(); } - @KafkaListener(id = "list6", topics = "annotated19", containerFactory = "batchManualFactory") + @KafkaListener(id = "list6", topics = "annotated19", containerFactory = "batchManualFactory2") public void listen15(List> 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 index b270b5ea..f4a375a5 100644 --- 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 class ConcurrentMessageListenerContainerTests { }; ConsumerFactory cf = mock(ConsumerFactory.class); Consumer 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>() { 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 index 6f170e5e..c1dd6290 100644 --- 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 class ContainerStoppingBatchErrorHandlerTests { 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 index e4601091..bcd8ad8a 100644 --- 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 class ContainerStoppingErrorHandlerBatchModeTests { 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 index a29353cd..96a5729c 100644 --- 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 class ContainerStoppingErrorHandlerRecordModeTests { 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 index 496bfed9..f4d2907b 100644 --- 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 @@ package org.springframework.kafka.listener; 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 class KafkaMessageListenerContainerTests { public void testRecordAckMock() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), isNull())).willReturn(consumer); + given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer); final Map>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), @@ -527,6 +528,7 @@ public class KafkaMessageListenerContainerTests { ).given(consumer).commitSync(any(Map.class)); containerProps.setMessageListener(messageListener); + containerProps.setClientId("clientId"); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); container.start(); @@ -546,7 +548,7 @@ public class KafkaMessageListenerContainerTests { public void testNonResponsiveConsumerEvent() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); - given(cf.createConsumer(isNull(), isNull())).willReturn(consumer); + given(cf.createConsumer(isNull(), eq(""), isNull())).willReturn(consumer); final Map>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), @@ -1060,7 +1062,8 @@ public class KafkaMessageListenerContainerTests { DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props) { @Override - public Consumer createConsumer(String groupId, String clientIdPrefix) { + public Consumer createConsumer(String groupId, String clientIdPrefix, + String clientIdSufffix) { return new KafkaConsumer(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 index 3e8408ae..9a153c28 100644 --- 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 class SeekToCurrentBatchErrorHandlerTests { 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 index 909e9ef4..9f2ce160 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java @@ -163,7 +163,7 @@ public class SeekToCurrentOnErrorBatchModeTXTests { 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/SeekToCurrentOnErrorBatchModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java index 5fa32cb6..90e05f55 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java @@ -139,7 +139,7 @@ public class SeekToCurrentOnErrorBatchModeTests { public ConsumerFactory consumerFactory() { ConsumerFactory consumerFactory = mock(ConsumerFactory.class); final Consumer consumer = consumer(); - given(consumerFactory.createConsumer(null, "-0")).willReturn(consumer); + given(consumerFactory.createConsumer(null, "", "-0")).willReturn(consumer); return consumerFactory; } 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 index f341fab7..d897005e 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java @@ -165,7 +165,7 @@ public class SeekToCurrentOnErrorRecordModeTXTests { 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/SeekToCurrentOnErrorRecordModeTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java index 79c45760..1585e5e6 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java @@ -143,7 +143,7 @@ public class SeekToCurrentOnErrorRecordModeTests { public ConsumerFactory consumerFactory() { ConsumerFactory consumerFactory = mock(ConsumerFactory.class); final Consumer consumer = consumer(); - given(consumerFactory.createConsumer(null, "-0")).willReturn(consumer); + given(consumerFactory.createConsumer(null, "", "-0")).willReturn(consumer); return consumerFactory; } 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 index c3346454..d5ca10f4 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java @@ -111,7 +111,7 @@ public class TransactionalContainerTests { } }).given(consumer).poll(anyLong()); ConsumerFactory cf = mock(ConsumerFactory.class); - willReturn(consumer).given(cf).createConsumer("group", null); + willReturn(consumer).given(cf).createConsumer("group", "", null); Producer producer = mock(Producer.class); final CountDownLatch closeLatch = new CountDownLatch(1); willAnswer(i -> { @@ -169,7 +169,7 @@ public class TransactionalContainerTests { } }).given(consumer).poll(anyLong()); ConsumerFactory cf = mock(ConsumerFactory.class); - willReturn(consumer).given(cf).createConsumer("group", null); + willReturn(consumer).given(cf).createConsumer("group", "", null); Producer producer = mock(Producer.class); final CountDownLatch closeLatch = new CountDownLatch(1); willAnswer(i -> { @@ -228,7 +228,7 @@ public class TransactionalContainerTests { } }).given(consumer).poll(anyLong()); ConsumerFactory cf = mock(ConsumerFactory.class); - willReturn(consumer).given(cf).createConsumer("group", null); + willReturn(consumer).given(cf).createConsumer("group", "", null); Producer producer = mock(Producer.class); final CountDownLatch closeLatch = new CountDownLatch(1); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 1dc16926..40146686 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -582,7 +582,7 @@ The `@KafkaListener` annotation provides a mechanism for simple POJO listeners: ---- public class Listener { - @KafkaListener(id = "foo", topics = "myTopic") + @KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId") public void listen(String data) { ... } @@ -628,6 +628,9 @@ public class KafkaConfig { Notice that to set container properties, you must use the `getContainerProperties()` method on the factory. It is used as a template for the actual properties injected into the container. +Starting with _version 2.1.1_, it is now possible to set the `client.id` property for consumers created by the annotation. +The `clientIdPrefix` is suffixed with `-n` where `n` is an integer representing the container number when using concurrency. + You can also configure POJO listeners with explicit topics and partitions (and, optionally, their initial offsets): [source, java] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index da8759ec..bdb59f6d 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -15,6 +15,13 @@ See <> for more information. Container Error handlers are now provided for both record and batch listeners that treat any exceptions thrown by the listener as fatal; they stop the container. See <> for more information. + +==== Client ID + +Starting with _version 2.1.1_, it is now possible to set the `client.id` prefix on `@KafkaListener`. +Previously, to customize the client id, you would need a separate consumer factory (and container factory) per listener. +The prefix is suffixed with `-n` to provide unique client ids when using concurrency. + ==== Migration Guide from 2.0 https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration].