@KafkaListener Consumer Property Overrides

Add the `properties` attribute to the `@KafkaListener` annotation to
enhance or override the consumer factory properties.

Author: garyrussell

.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

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

@@ -220,4 +220,25 @@
 	 */
 	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 {};
+
 }

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.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 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 		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());

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

@@ -296,8 +296,6 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 
 		endpoint.setupListenerContainer(instance, this.messageConverter);
 		initializeContainer(instance, endpoint);
-		instance.getContainerProperties().setGroupId(endpoint.getGroupId());
-		instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());
 
 		return instance;
 	}
@@ -371,6 +369,11 @@ else if (this.autoStartup != null) {
 		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

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

@@ -20,6 +20,7 @@
 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.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 void setReplyHeadersConfigurer(ReplyHeadersConfigurer replyHeadersConfigu
 		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();

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.

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 @@ default Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String
 	 * 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 @@ default Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String
 	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.

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 @@
 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,20 +99,34 @@ public Deserializer<V> getValueDeserializer() {
 	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 Properties properties) {
+
 		boolean overrideClientIdPrefix = StringUtils.hasText(clientIdPrefix);
 		String clientIdSuffix = clientIdSuffixArg;
 		if (clientIdSuffix == null) {
 			clientIdSuffix = "";
 		}
 		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 @@ protected KafkaConsumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nul
 						(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);
 		}
 	}

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

@@ -18,6 +18,7 @@
 
 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.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 enum AckMode {
 
 	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 void setMissingTopicsFatal(boolean missingTopicsFatal) {
 		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 ["

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

@@ -497,7 +497,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 					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);