diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 00000000..26c1be09 --- /dev/null +++ b/.editorconfig @@ -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 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 0c7e5807..7eeea409 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 @@ -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). + *

Supported Syntax

+ *

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}: + *

+ * {@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 {}; + } 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 019c26c1..7ac0760a 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 @@ -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 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()); 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 2558c2fe..15f27a73 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 @@ -296,8 +296,6 @@ public abstract class AbstractKafkaListenerContainerFactory 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 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(); 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 aa4c351a..103c6df3 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 @@ -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. 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 52cc22a8..01770c98 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 @@ -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 { * 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 { Consumer 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 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. 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 3afd2f0a..83ea6886 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 @@ -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 implements ConsumerFactory public Consumer createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix, @Nullable String clientIdSuffix) { - return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix); + return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffix, null); + } + + @Override + public Consumer createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix, + @Nullable final String clientIdSuffixArg, @Nullable Properties properties) { + + return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties); + } + + @Deprecated + protected KafkaConsumer createKafkaConsumer(@Nullable String groupId, @Nullable String clientIdPrefix, + @Nullable final String clientIdSuffixArg) { + + return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, null); } protected KafkaConsumer 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 implements ConsumerFactory } 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 implements ConsumerFactory (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); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java index 96ca65d1..cd6e5271 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java @@ -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 [" 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 eb51e4d5..048ca791 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 @@ -497,7 +497,8 @@ public class KafkaMessageListenerContainer // 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); 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 40670f92..783fdabf 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 @@ -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 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(); 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 65be32e9..4b8f4429 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 @@ -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 cf = mock(ConsumerFactory.class); Consumer 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>() { 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 81a6d4ed..f36f8421 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 @@ -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; } 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 d8450537..6b7727b8 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 @@ -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; } 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 80059e31..3f8c144f 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 @@ -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; } 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 22462809..2d2e45b2 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 @@ -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 cf = mock(ConsumerFactory.class); Consumer 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>> 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 cf = mock(ConsumerFactory.class); Consumer 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>> 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 cf = mock(ConsumerFactory.class); Consumer 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>> 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 cf = mock(ConsumerFactory.class); Consumer 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 createConsumer(String groupId, String clientIdPrefix, - String clientIdSuffix) { + String clientIdSuffix, Properties properties) { return new KafkaConsumer(props) { @Override @@ -1854,7 +1855,7 @@ public class KafkaMessageListenerContainerTests { public void testPauseResume() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer 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>> 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 cf = mock(ConsumerFactory.class); Consumer 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 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 cf = mock(ConsumerFactory.class); Consumer 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>> records1 = new HashMap<>(); records1.put(topicPartition, Arrays.asList( @@ -2099,7 +2100,7 @@ public class KafkaMessageListenerContainerTests { public void testCommitErrorHandlerCalled() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer 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>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java index 3611427f..e35c1a63 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java @@ -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; } 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 e3b211ff..b7894dbe 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 @@ -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; } 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 a56a852f..ced4939a 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 @@ -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; } 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 dfd22237..00f89473 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 @@ -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; } 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 8bdbfd27..a7068c0e 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 @@ -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; } 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 aba53068..03f393d9 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 @@ -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; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java index 382bff8a..b26b65f3 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TestOOMError.java @@ -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 cf = mock(ConsumerFactory.class); Consumer 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>> 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 cf = mock(ConsumerFactory.class); Consumer 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>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), 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 640f9361..290c43c9 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 @@ -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); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 3e0bbbed..622358f2 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -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 `>` 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`. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 59ce2f2a..5bc01eaf 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -57,6 +57,9 @@ See <> for more information. It is now easier to configure a `Validator` for `@Payload` validation. See <> 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 <> for more information. + ==== Header Mapping Changes Headers of type `MimeType` and `MediaType` are now mapped as simple strings in the `RecordHeader` value.