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

@KafkaListener Consumer Property Overrides

Browse Source 
Add the `properties` attribute to the `@KafkaListener` annotation to
enhance or override the consumer factory properties.
This commit is contained in:
Gary Russell
2019-02-07 16:19:25 -05:00
committed by Artem Bilan
parent c5b2ee47cb
commit ff174e0a95
26 changed files with 245 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
.editorconfig Normal file
View File

@@ -0,0 +1,11 @@
root=true
[*.java]
indent_style = tab
indent_size = 4
continuation_indent_size = 8
[*.xml]
indent_style = tab
indent_size = 4
continuation_indent_size = 8

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

@@ -220,4 +220,25 @@ public @interface KafkaListener {
*/
String autoStartup() default "";
/**
* Kafka consumer properties; they will supersede any properties with the same name
* defined in the consumer factory (if the consumer factory supports property overrides).
* <h3>Supported Syntax</h3>
* <p>The supported syntax for key-value pairs is the same as the
* syntax defined for entries in a Java
* {@linkplain java.util.Properties#load(java.io.Reader) properties file}:
* <ul>
* <li>{@code key=value}</li>
* <li>{@code key:value}</li>
* <li>{@code key value}</li>
* </ul>
* {@code group.id} and {@code client.id} are ignored.
* @return the properties.
* @since 2.2.4
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #groupId()
* @see #clientIdPrefix()
*/
String[] properties() default {};
}

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

@@ -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.
@@ -16,6 +16,8 @@
package org.springframework.kafka.annotation;
import java.io.IOException;
import java.io.StringReader;
import java.lang.reflect.Method;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
@@ -27,6 +29,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@@ -439,6 +442,19 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
if (StringUtils.hasText(autoStartup)) {
endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));
}
String[] propertyStrings = kafkaListener.properties();
if (propertyStrings.length > 0) {
Properties properties = new Properties();
for (String property : propertyStrings) {
try {
properties.load(new StringReader(resolveExpressionAsString(property, "property")));
}
catch (IOException e) {
this.logger.error("Failed to load property " + property + ", continuing...", e);
}
}
endpoint.setConsumerProperties(properties);
}
KafkaListenerContainerFactory<?> factory = null;
String containerFactoryBeanName = resolve(kafkaListener.containerFactory());

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

@@ -296,8 +296,6 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
endpoint.setupListenerContainer(instance, this.messageConverter);
initializeContainer(instance, endpoint);
instance.getContainerProperties().setGroupId(endpoint.getGroupId());
instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());
return instance;
}
@@ -371,6 +369,11 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
if (this.applicationEventPublisher != null) {
instance.setApplicationEventPublisher(this.applicationEventPublisher);
}
instance.getContainerProperties().setGroupId(endpoint.getGroupId());
instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());
if (endpoint.getConsumerProperties() != null) {
instance.getContainerProperties().setConsumerProperties(endpoint.getConsumerProperties());
}
}
@Override

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

@@ -20,6 +20,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Properties;
import java.util.regex.Pattern;
import org.apache.commons.logging.Log;
@@ -46,6 +47,7 @@ import org.springframework.kafka.listener.adapter.ReplyHeadersConfigurer;
import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.converter.MessageConverter;
import org.springframework.lang.Nullable;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
@@ -109,6 +111,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
private ReplyHeadersConfigurer replyHeadersConfigurer;
private Properties consumerProperties;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
@@ -386,6 +390,27 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
this.replyHeadersConfigurer = replyHeadersConfigurer;
}
@Override
@Nullable
public Properties getConsumerProperties() {
return this.consumerProperties;
}
/**
* Set the consumer properties that will be merged with the consumer properties
* provided by the consumer factory; properties here will supersede any with the same
* name(s) in the consumer factory.
* {@code group.id} and {@code client.id} are ignored.
* @param consumerProperties the properties.
* @since 2.1.4
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #setGroupId(String)
* @see #setClientIdPrefix(String)
*/
public void setConsumerProperties(Properties consumerProperties) {
this.consumerProperties = consumerProperties;
}
@Override
public void afterPropertiesSet() {
boolean topicsEmpty = getTopics().isEmpty();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-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.
@@ -17,11 +17,13 @@
package org.springframework.kafka.config;
import java.util.Collection;
import java.util.Properties;
import java.util.regex.Pattern;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.converter.MessageConverter;
import org.springframework.lang.Nullable;
/**
* Model for a Kafka listener endpoint. Can be used against a
@@ -97,6 +99,22 @@ public interface KafkaListenerEndpoint {
*/
Boolean getAutoStartup();
/**
* Get the consumer properties that will be merged with the consumer properties
* provided by the consumer factory; properties here will supersede any with the same
* name(s) in the consumer factory.
* {@code group.id} and {@code client.id} are ignored.
* @return the properties.
* @since 2.1.4
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #getGroupId()
* @see #getClientIdPrefix()
*/
@Nullable
default Properties getConsumerProperties() {
return null;
}
/**
* Setup the specified message listener container with the model
* defined by this endpoint.

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

@@ -17,6 +17,7 @@
package org.springframework.kafka.core;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.serialization.Deserializer;
@@ -70,8 +71,6 @@ public interface ConsumerFactory<K, V> {
* 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.
* @param groupId the group id.
* @param clientIdPrefix the prefix.
* @param clientIdSuffix the suffix.
@@ -81,6 +80,24 @@ public interface ConsumerFactory<K, V> {
Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
@Nullable 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. In addition, consumer properties can
* be overridden if the factory implementation supports it.
* @param groupId the group id.
* @param clientIdPrefix the prefix.
* @param clientIdSuffix the suffix.
* @param properties the properties to override.
* @return the consumer.
* @since 2.2.4
*/
default Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
@Nullable String clientIdSuffix, @Nullable Properties properties) {
return createConsumer(groupId, clientIdPrefix, clientIdSuffix);
}
/**
* Return true if consumers created by this factory use auto commit.
* @return true if auto commit.

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

@@ -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.
@@ -19,6 +19,7 @@ package org.springframework.kafka.core;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -98,11 +99,25 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
@Nullable String clientIdSuffix) {
return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix);
return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix, null);
}
@Override
public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
@Nullable final String clientIdSuffixArg, @Nullable Properties properties) {
return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties);
}
@Deprecated
protected KafkaConsumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
@Nullable final String clientIdSuffixArg) {
return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, null);
}
protected KafkaConsumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
@Nullable final String clientIdSuffixArg) {
@Nullable final String clientIdSuffixArg, @Nullable Properties properties) {
boolean overrideClientIdPrefix = StringUtils.hasText(clientIdPrefix);
String clientIdSuffix = clientIdSuffixArg;
@@ -111,7 +126,7 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
}
boolean shouldModifyClientId = (this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)
&& StringUtils.hasText(clientIdSuffix)) || overrideClientIdPrefix;
if (groupId == null && !shouldModifyClientId) {
if (groupId == null && properties == null && !shouldModifyClientId) {
return createKafkaConsumer(this.configs);
}
else {
@@ -124,6 +139,13 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
(overrideClientIdPrefix ? clientIdPrefix
: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
}
if (properties != null) {
properties.forEach((k, v) -> {
if (!k.equals(ConsumerConfig.CLIENT_ID_CONFIG) && !k.equals(ConsumerConfig.GROUP_ID_CONFIG)) {
modifiedConfigs.put((String) k, v);
}
});
}
return createKafkaConsumer(modifiedConfigs);
}
}

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

@@ -18,6 +18,7 @@ package org.springframework.kafka.listener;
import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.Properties;
import java.util.regex.Pattern;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -26,6 +27,7 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.kafka.support.LogIfLevelEnabled;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
@@ -220,6 +222,8 @@ public class ContainerProperties {
private boolean missingTopicsFatal = true;
private Properties consumerProperties;
/**
* Create properties for a container that will subscribe to the specified topics.
* @param topics the topics.
@@ -609,6 +613,37 @@ public class ContainerProperties {
this.missingTopicsFatal = missingTopicsFatal;
}
/**
* Get the consumer properties that will be merged with the consumer properties
* provided by the consumer factory; properties here will supersede any with the same
* name(s) in the consumer factory.
* {@code group.id} and {@code client.id} are ignored.
* @return the properties.
* @since 2.1.4
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #setGroupId(String)
* @see #setClientId(String)
*/
@Nullable
public Properties getConsumerProperties() {
return this.consumerProperties;
}
/**
* Set the consumer properties that will be merged with the consumer properties
* provided by the consumer factory; properties here will supersede any with the same
* name(s) in the consumer factory.
* {@code group.id} and {@code client.id} are ignored.
* @param consumerProperties the properties.
* @since 2.1.4
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #setGroupId(String)
* @see #setClientId(String)
*/
public void setConsumerProperties(Properties consumerProperties) {
this.consumerProperties = consumerProperties;
}
@Override
public String toString() {
return "ContainerProperties ["

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

@@ -497,7 +497,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
this.consumerGroupId,
this.containerProperties.getClientId(),
KafkaMessageListenerContainer.this.clientIdSuffix);
KafkaMessageListenerContainer.this.clientIdSuffix,
this.containerProperties.getConsumerProperties());
if (this.transactionManager != null) {
this.transactionTemplate = new TransactionTemplate(this.transactionManager);

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

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.BDDMockito.willThrow;
@@ -252,6 +253,9 @@ public class EnableKafkaIntegrationTests {
assertThat(this.listener.listen4Consumer).isSameAs(KafkaTestUtils.getPropertyValue(KafkaTestUtils
.getPropertyValue(this.registry.getListenerContainer("qux"), "containers", List.class).get(0),
"listenerConsumer.consumer"));
assertThat(
KafkaTestUtils.getPropertyValue(this.listener.listen4Consumer, "fetcher.maxPollRecords", Integer.class))
.isEqualTo(100);
assertThat(this.quxGroup).hasSize(1);
assertThat(this.quxGroup.get(0)).isSameAs(manualContainer);
List<?> containers = KafkaTestUtils.getPropertyValue(manualContainer, "containers", List.class);
@@ -878,7 +882,7 @@ public class EnableKafkaIntegrationTests {
willAnswer(i -> {
Consumer<Integer, CharSequence> spy =
spy(consumerFactory().createConsumer(i.getArgument(0), i.getArgument(1),
i.getArgument(2)));
i.getArgument(2), i.getArgument(3)));
willAnswer(invocation -> {
try {
@@ -890,7 +894,7 @@ public class EnableKafkaIntegrationTests {
}).given(spy).commitSync(anyMap());
return spy;
}).given(spiedCf).createConsumer(anyString(), anyString(), anyString());
}).given(spiedCf).createConsumer(anyString(), anyString(), anyString(), isNull());
factory.setConsumerFactory(spiedCf);
factory.setBatchListener(true);
factory.setRecordFilterStrategy(recordFilter());
@@ -1413,7 +1417,10 @@ public class EnableKafkaIntegrationTests {
}
@KafkaListener(id = "qux", topics = "annotated4", containerFactory = "kafkaManualAckListenerContainerFactory",
containerGroup = "qux#{'Group'}")
containerGroup = "qux#{'Group'}", properties = {
"max.poll.interval.ms:#{'${poll.interval:60000}'}",
ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=#{'${poll.recs:100}'}"
})
public void listen4(@Payload String foo, Acknowledgment ack, Consumer<?, ?> consumer) {
this.ack = ack;
this.ack.acknowledge();

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

@@ -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.
@@ -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.anyString;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
@@ -423,7 +424,7 @@ public class ConcurrentMessageListenerContainerTests {
};
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(anyString(), anyString(), anyString())).willReturn(consumer);
given(cf.createConsumer(anyString(), anyString(), anyString(), isNull())).willReturn(consumer);
given(consumer.poll(any(Duration.class)))
.willAnswer(new Answer<ConsumerRecords<Integer, String>>() {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -123,7 +123,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", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -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", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -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", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -40,6 +40,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -530,7 +531,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(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), 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"),
@@ -597,7 +598,7 @@ public class KafkaMessageListenerContainerTests {
private void testRecordAckMockForeignThreadGuts(AckMode ackMode) throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), 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"),
@@ -660,7 +661,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(eq("grp"), eq(""), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq(""), isNull(), 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"),
@@ -700,7 +701,7 @@ public class KafkaMessageListenerContainerTests {
public void testNonResponsiveConsumerEventNotIssuedWithActiveConsumer() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(isNull(), eq(""), isNull())).willReturn(consumer);
given(cf.createConsumer(isNull(), eq(""), isNull(), isNull())).willReturn(consumer);
ConsumerRecords records = new ConsumerRecords(Collections.emptyMap());
CountDownLatch latch = new CountDownLatch(20);
given(consumer.poll(any(Duration.class))).willAnswer(i -> {
@@ -1210,7 +1211,7 @@ public class KafkaMessageListenerContainerTests {
@Override
public Consumer<Integer, String> createConsumer(String groupId, String clientIdPrefix,
String clientIdSuffix) {
String clientIdSuffix, Properties properties) {
return new KafkaConsumer<Integer, String>(props) {
@Override
@@ -1854,7 +1855,7 @@ public class KafkaMessageListenerContainerTests {
public void testPauseResume() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), 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"),
@@ -1921,7 +1922,7 @@ public class KafkaMessageListenerContainerTests {
public void testInitialSeek() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), isNull())).willReturn(consumer);
ConsumerRecords<Integer, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
final CountDownLatch latch = new CountDownLatch(1);
given(consumer.poll(any(Duration.class))).willAnswer(i -> {
@@ -2033,7 +2034,7 @@ public class KafkaMessageListenerContainerTests {
public void testAckModeCount() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), isNull())).willReturn(consumer);
TopicPartition topicPartition = new TopicPartition("foo", 0);
final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records1 = new HashMap<>();
records1.put(topicPartition, Arrays.asList(
@@ -2099,7 +2100,7 @@ public class KafkaMessageListenerContainerTests {
public void testCommitErrorHandlerCalled() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), 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"),

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

@@ -132,7 +132,7 @@ public class RemainingRecordsErrorHandlerTests {
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", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -143,7 +143,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", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -164,7 +164,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", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -140,7 +140,7 @@ public class SeekToCurrentOnErrorBatchModeTests {
public ConsumerFactory consumerFactory() {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
final Consumer consumer = consumer();
given(consumerFactory.createConsumer("grp", "", "-0")).willReturn(consumer);
given(consumerFactory.createConsumer("grp", "", "-0", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -166,7 +166,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", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -144,7 +144,7 @@ public class SeekToCurrentOnErrorRecordModeTests {
public ConsumerFactory consumerFactory() {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
final Consumer consumer = consumer();
given(consumerFactory.createConsumer("grp", "", "-0")).willReturn(consumer);
given(consumerFactory.createConsumer("grp", "", "-0", null)).willReturn(consumer);
return consumerFactory;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2018 the original author or authors.
* Copyright 2018-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.
@@ -54,7 +54,7 @@ public class TestOOMError {
public void testOOMCMLC() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), eq("-0"))).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), eq("-0"), 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"),
@@ -93,7 +93,7 @@ public class TestOOMError {
public void testOOMKMLC() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull())).willReturn(consumer);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), 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"),

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-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.
@@ -154,7 +154,7 @@ public class TransactionalContainerTests {
}
}).given(consumer).poll(any(Duration.class));
ConsumerFactory cf = mock(ConsumerFactory.class);
willReturn(consumer).given(cf).createConsumer("group", "", null);
willReturn(consumer).given(cf).createConsumer("group", "", null, null);
Producer producer = mock(Producer.class);
final CountDownLatch closeLatch = new CountDownLatch(2);
willAnswer(i -> {
@@ -237,7 +237,7 @@ public class TransactionalContainerTests {
return null;
}).given(consumer).seek(any(), anyLong());
ConsumerFactory cf = mock(ConsumerFactory.class);
willReturn(consumer).given(cf).createConsumer("group", "", null);
willReturn(consumer).given(cf).createConsumer("group", "", null, null);
Producer producer = mock(Producer.class);
final CountDownLatch closeLatch = new CountDownLatch(1);
willAnswer(i -> {
@@ -304,7 +304,7 @@ public class TransactionalContainerTests {
return null;
}).given(consumer).seek(any(), anyLong());
ConsumerFactory cf = mock(ConsumerFactory.class);
willReturn(consumer).given(cf).createConsumer("group", "", null);
willReturn(consumer).given(cf).createConsumer("group", "", null, null);
Producer producer = mock(Producer.class);
final CountDownLatch closeLatch = new CountDownLatch(1);
willAnswer(i -> {
@@ -368,7 +368,7 @@ public class TransactionalContainerTests {
}
}).given(consumer).poll(any(Duration.class));
ConsumerFactory cf = mock(ConsumerFactory.class);
willReturn(consumer).given(cf).createConsumer("group", "", null);
willReturn(consumer).given(cf).createConsumer("group", "", null, null);
Producer producer = mock(Producer.class);
final CountDownLatch closeLatch = new CountDownLatch(1);

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

@@ -1103,6 +1103,7 @@ public void pollResults(ConsumerRecords<?, ?> records) {
IMPORTANT: If the container factory has a `RecordFilterStrategy` configured, it is ignored for `ConsumerRecords<?, ?>` listeners, with a `WARN` log message emitted.
Records can only be filtered with a batch listener if the `<List<?>>` form of listener is used.
[[annotation-properties]]
====== Annotation Properties
Starting with version 2.0, the `id` property (if present) is used as the Kafka consumer `group.id` property, overriding the configured property in the consumer factory, if present.
@@ -1178,6 +1179,20 @@ The following example shows how to do so:
----
====
Starting with version 2.2.4, you can specify Kafka consumer properties directly on the annotation, these will override any properties with the same name configured in the consumer factory. You **cannot** specify the `group.id` and `client.id` properties this way; they will be ignored; use the `groupId` and `clientIdPrefix` annotation properties for those.
The properties are specified as individual strings with the normal Java `Properties` file format: `foo:bar`, `foo=bar`, or `foo bar`.
====
[source, java]
----
@KafkaListener(topics = "myTopic", groupId="group", properties= {
"max.poll.interval.ms:60000",
ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=100"
})
----
====
===== Container Thread Naming
Listener containers currently use two task executors, one to invoke the consumer and another that is used to invoke the listener when the kafka consumer property `enable.auto.commit` is `false`.

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

@@ -57,6 +57,9 @@ See <<kafka-listener-meta>> for more information.
It is now easier to configure a `Validator` for `@Payload` validation.
See <<kafka-validation>> for more information.
You can now specify kafka consumer properties directly on the annotation; these will override any properties with the same name defined in the consumer factory (since version 2.2.4).
See <<annotation-properties>> for more information.
==== Header Mapping Changes
Headers of type `MimeType` and `MediaType` are now mapped as simple strings in the `RecordHeader` value.