Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-518: Add clientIdPrefix to @KafkaListener

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/518

* Polishing - PR Comments
This commit is contained in:
Gary Russell
2017-12-20 13:03:53 -05:00
committed by Artem Bilan
parent 7d1aa4eadc
commit a39780cae3
23 changed files with 163 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListener.java
View File

@@ -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 "";
}

2
spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
View File

@@ -383,6 +383,8 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
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);

1
spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java
View File

@@ -227,6 +227,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
endpoint.setupListenerContainer(instance, this.messageConverter);
initializeContainer(instance);
instance.getContainerProperties().setGroupId(endpoint.getGroupId());
instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());
return instance;
}

17
spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java
View File

@@ -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 abstract class AbstractKafkaListenerEndpoint<K, V>
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();

9
spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java
View File

@@ -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.

17
spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java
View File

@@ -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.

23
spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
View File

@@ -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<K, V> implements ConsumerFactory<K, V>
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 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
}
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);
}

7
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -387,8 +387,11 @@ public class KafkaMessageListenerContainer<K, V> 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<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);

23
spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java
View File

@@ -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;
}
}

39
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -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<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 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<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 class EnableKafkaIntegrationTests {
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 class EnableKafkaIntegrationTests {
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 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<Message<?>> list, Acknowledgment ack) {
this.payload = list;
this.ack = ack;

2
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -393,7 +393,7 @@ public class ConcurrentMessageListenerContainerTests {
};
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>>() {

2
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java
View File

@@ -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;
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java
View File

@@ -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;
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java
View File

@@ -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;
}

9
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -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<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 class KafkaMessageListenerContainerTests {
).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 class KafkaMessageListenerContainerTests {
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 class KafkaMessageListenerContainerTests {
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

2
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java
View File

@@ -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;
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTXTests.java
View File

@@ -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;
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorBatchModeTests.java
View File

@@ -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;
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTXTests.java
View File

@@ -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;
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java
View File

@@ -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;
}

6
spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java
View File

@@ -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);

5
src/reference/asciidoc/kafka.adoc
View File

@@ -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]

7
src/reference/asciidoc/whats-new.adoc
View File

@@ -15,6 +15,13 @@ See <<serdes>> 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 <<annotation-error-handling>> 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].