Use ConsumerProperties in KafkaMessageSource

* Preserve existing constructors

* Add @deprecated in java docs

Author: Walliee

src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java

@@ -30,6 +30,7 @@
  * Parser for the inbound channel adapter.
  *
  * @author Gary Russell
+ * @author Anshul Mehra
  * @since 3.2
  *
  */
@@ -44,6 +45,10 @@ protected boolean shouldGenerateIdAsFallback() {
 	protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(KafkaMessageSource.class);
 		builder.addConstructorArgReference(element.getAttribute("consumer-factory"));
+		boolean hasConsumerProperties = StringUtils.hasText(element.getAttribute("consumer-properties"));
+		if (hasConsumerProperties) {
+			builder.addConstructorArgReference(element.getAttribute("consumer-properties"));
+		}
 		String attribute = element.getAttribute("ack-factory");
 		if (StringUtils.hasText(attribute)) {
 			builder.addConstructorArgReference(attribute);
@@ -52,13 +57,15 @@ protected BeanMetadataElement parseSource(Element element, ParserContext parserC
 		if (StringUtils.hasText(attribute)) {
 			builder.addConstructorArgValue(attribute);
 		}
-		builder.addConstructorArgValue(element.getAttribute("topics"));
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "client-id");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "group-id");
+		if (!hasConsumerProperties) {
+			builder.addConstructorArgValue(element.getAttribute("topics"));
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "client-id");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "group-id");
+			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "rebalance-listener");
+		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "raw-header", "rawMessageHeader");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "rebalance-listener");
 		return builder.getBeanDefinition();
 	}
 

src/main/java/org/springframework/integration/kafka/dsl/Kafka.java

@@ -27,6 +27,7 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.listener.ConsumerProperties;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.GenericMessageListenerContainer;
 import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
@@ -38,6 +39,7 @@
  * @author Artem Bilan
  * @author Nasko Vasilev
  * @author Gary Russell
+ * @author Anshul Mehra
  *
  * @since 3.0
  */
@@ -79,67 +81,133 @@ public static <K, V> KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandler
 	 * @param <V> the Kafka message value type.
 	 * @return the spec.
 	 * @since 3.0.1
+	 * @deprecated in favor of {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties)}
 	 */
+	@Deprecated
 	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
 			ConsumerFactory<K, V> consumerFactory, String... topics) {
 
-		return inboundChannelAdapter(consumerFactory, false, topics);
+		return inboundChannelAdapter(consumerFactory, new ConsumerProperties(topics), false);
 	}
 
 	/**
 	 * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and
-	 * topics.
+	 * topics with a custom ack callback factory.
 	 * @param consumerFactory the consumer factory.
-	 * @param allowMultiFetch true to fetch multiple records on each poll.
+	 * @param ackCallbackFactory the callback factory.
 	 * @param topics the topic(s).
 	 * @param <K> the Kafka message key type.
 	 * @param <V> the Kafka message value type.
 	 * @return the spec.
-	 * @since 3.2
+	 * @since 3.0.1
+	 * @deprecated in favor of
+	 * {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)}
 	 */
+	@Deprecated
 	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
-			ConsumerFactory<K, V> consumerFactory, boolean allowMultiFetch, String... topics) {
+			ConsumerFactory<K, V> consumerFactory,
+			KafkaAckCallbackFactory<K, V> ackCallbackFactory, String... topics) {
 
-		return new KafkaInboundChannelAdapterSpec<>(consumerFactory, allowMultiFetch, topics);
+		return inboundChannelAdapter(consumerFactory, ackCallbackFactory, false, topics);
 	}
 
 	/**
 	 * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and
 	 * topics with a custom ack callback factory.
 	 * @param consumerFactory the consumer factory.
 	 * @param ackCallbackFactory the callback factory.
+	 * @param allowMultiFetch true to fetch multiple records on each poll.
 	 * @param topics the topic(s).
 	 * @param <K> the Kafka message key type.
 	 * @param <V> the Kafka message value type.
 	 * @return the spec.
 	 * @since 3.0.1
+	 * @deprecated in favor of
+	 * {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)}
 	 */
+	@Deprecated
 	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
 			ConsumerFactory<K, V> consumerFactory,
-			KafkaAckCallbackFactory<K, V> ackCallbackFactory, String... topics) {
+			KafkaAckCallbackFactory<K, V> ackCallbackFactory,
+			boolean allowMultiFetch,
+			String... topics) {
 
-		return inboundChannelAdapter(consumerFactory, ackCallbackFactory, false, topics);
+		return new KafkaInboundChannelAdapterSpec<>(consumerFactory, new ConsumerProperties(topics), ackCallbackFactory, allowMultiFetch);
+	}
+
+	/**
+	 * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and
+	 * topics.
+	 * @param consumerFactory the consumer factory.
+	 * @param consumerProperties the consumerProperties.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the spec.
+	 * @since 3.2
+	 */
+	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
+			ConsumerFactory<K, V> consumerFactory, ConsumerProperties consumerProperties) {
+
+		return inboundChannelAdapter(consumerFactory, consumerProperties, false);
+	}
+
+	/**
+	 * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and
+	 * topics.
+	 * @param consumerFactory the consumer factory.
+	 * @param consumerProperties the consumerProperties.
+	 * @param allowMultiFetch true to fetch multiple records on each poll.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the spec.
+	 * @since 3.2
+	 */
+	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
+			ConsumerFactory<K, V> consumerFactory,
+			ConsumerProperties consumerProperties,
+			boolean allowMultiFetch) {
+
+		return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, allowMultiFetch);
+	}
+
+	/**
+	 * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and
+	 * topics with a custom ack callback factory.
+	 * @param consumerFactory the consumer factory.
+	 * @param consumerProperties the consumerProperties.
+	 * @param ackCallbackFactory the callback factory.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the spec.
+	 * @since 3.2
+	 */
+	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
+			ConsumerFactory<K, V> consumerFactory,
+			ConsumerProperties consumerProperties,
+			KafkaAckCallbackFactory<K, V> ackCallbackFactory) {
+
+		return inboundChannelAdapter(consumerFactory, consumerProperties, ackCallbackFactory, false);
 	}
 
 	/**
 	 * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and
 	 * topics with a custom ack callback factory.
 	 * @param consumerFactory the consumer factory.
+	 * @param consumerProperties the consumerProperties.
 	 * @param ackCallbackFactory the callback factory.
 	 * @param allowMultiFetch true to fetch multiple records on each poll.
-	 * @param topics the topic(s).
 	 * @param <K> the Kafka message key type.
 	 * @param <V> the Kafka message value type.
 	 * @return the spec.
-	 * @since 3.0.1
+	 * @since 3.2
 	 */
 	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
 			ConsumerFactory<K, V> consumerFactory,
+			ConsumerProperties consumerProperties,
 			KafkaAckCallbackFactory<K, V> ackCallbackFactory,
-			boolean allowMultiFetch,
-			String... topics) {
+			boolean allowMultiFetch) {
 
-		return new KafkaInboundChannelAdapterSpec<>(consumerFactory, ackCallbackFactory, allowMultiFetch, topics);
+		return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, ackCallbackFactory, allowMultiFetch);
 	}
 
 	/**

src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java

@@ -17,11 +17,16 @@
 package org.springframework.integration.kafka.dsl;
 
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.dsl.MessageSourceSpec;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckCallbackFactory;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.listener.ConsumerProperties;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 
 /**
@@ -31,53 +36,155 @@
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Anshul Mehra
  *
  * @since 3.0.1
  *
  */
 public class KafkaInboundChannelAdapterSpec<K, V>
 		extends MessageSourceSpec<KafkaInboundChannelAdapterSpec<K, V>, KafkaMessageSource<K, V>> {
 
+	/**
+	 * Create an initial {@link KafkaMessageSource} with the consumer factory and
+	 * topics.
+	 * @param consumerFactory the consumer factory.
+	 * @param allowMultiFetch true to allow {@code max.poll.records > 1}.
+	 * @param topics the topics.
+	 * @deprecated in favor of
+	 * {@link #KafkaInboundChannelAdapterSpec(ConsumerFactory, ConsumerProperties, boolean)}
+	 */
+	@Deprecated
 	KafkaInboundChannelAdapterSpec(ConsumerFactory<K, V> consumerFactory, boolean allowMultiFetch, String... topics) {
-		this.target = new KafkaMessageSource<>(consumerFactory, allowMultiFetch, topics);
+		this.target = new KafkaMessageSource<>(consumerFactory, new ConsumerProperties(topics), allowMultiFetch);
 	}
 
+	/**
+	 * Create an initial {@link KafkaMessageSource} with the consumer factory and
+	 * topics with a custom ack callback factory.
+	 * @param consumerFactory the consumer factory.
+	 * @param ackCallbackFactory the callback factory.
+	 * @param allowMultiFetch true to allow {@code max.poll.records > 1}.
+	 * @param topics the topics.
+	 * @deprecated in favor of
+	 * {@link #KafkaInboundChannelAdapterSpec(ConsumerFactory, ConsumerProperties,
+	 * KafkaAckCallbackFactory, boolean)}
+	 */
+	@Deprecated
 	KafkaInboundChannelAdapterSpec(ConsumerFactory<K, V> consumerFactory,
 			KafkaAckCallbackFactory<K, V> ackCallbackFactory, boolean allowMultiFetch, String... topics) {
 
-		this.target = new KafkaMessageSource<>(consumerFactory, ackCallbackFactory, allowMultiFetch, topics);
+		this.target = new KafkaMessageSource<>(consumerFactory, new ConsumerProperties(topics), ackCallbackFactory, allowMultiFetch);
 	}
 
+	/**
+	 * Create an initial {@link KafkaMessageSource} with the consumer factory and
+	 * topics with a custom ack callback factory.
+	 * @param consumerFactory the consumer factory.
+	 * @param consumerProperties the consumer properties.
+	 * @param allowMultiFetch true to allow {@code max.poll.records > 1}.
+	 */
+	KafkaInboundChannelAdapterSpec(ConsumerFactory<K, V> consumerFactory,
+			ConsumerProperties consumerProperties, boolean allowMultiFetch) {
+		this.target = new KafkaMessageSource<>(consumerFactory, consumerProperties, allowMultiFetch);
+	}
+
+	/**
+	 * Create an initial {@link KafkaMessageSource} with the consumer factory and
+	 * topics with a custom ack callback factory.
+	 * @param consumerFactory the consumer factory.
+	 * @param consumerProperties the consumer properties.
+	 * @param ackCallbackFactory the callback factory.
+	 * @param allowMultiFetch true to allow {@code max.poll.records > 1}.
+	 */
+	KafkaInboundChannelAdapterSpec(ConsumerFactory<K, V> consumerFactory,
+			ConsumerProperties consumerProperties,
+			KafkaAckCallbackFactory<K, V> ackCallbackFactory, boolean allowMultiFetch) {
+
+		this.target = new KafkaMessageSource<>(consumerFactory, consumerProperties, ackCallbackFactory, allowMultiFetch);
+	}
+
+	/**
+	 * Set the group.id property for the consumer.
+	 * @param groupId the group id.
+	 * @return the spec.
+	 * @see ConsumerProperties
+	 * @deprecated in favor of using {@link ConsumerProperties}
+	 */
+	@Deprecated
 	public KafkaInboundChannelAdapterSpec<K, V> groupId(String groupId) {
 		this.target.setGroupId(groupId);
 		return this;
 	}
 
+	/**
+	 * Set the client.id property for the consumer.
+	 * @param clientId the client id.
+	 * @return the spec.
+	 * @see ConsumerProperties
+	 * @deprecated in favor of using {@link ConsumerProperties}
+	 */
+	@Deprecated
 	public KafkaInboundChannelAdapterSpec<K, V> clientId(String clientId) {
 		this.target.setClientId(clientId);
 		return this;
 	}
 
+	/**
+	 * Set the pollTimeout for the poll() operations.
+	 * @param pollTimeout the poll timeout.
+	 * @return the spec.
+	 * @see ConsumerProperties
+	 * @deprecated in favor of using {@link ConsumerProperties}
+	 */
+	@Deprecated
 	public KafkaInboundChannelAdapterSpec<K, V> pollTimeout(long pollTimeout) {
 		this.target.setPollTimeout(pollTimeout);
 		return this;
 	}
 
+	/**
+	 * Set the message converter to replace the default.
+	 * {@link MessagingMessageConverter}.
+	 * @param messageConverter the converter.
+	 * @return the spec.
+	 */
 	public KafkaInboundChannelAdapterSpec<K, V> messageConverter(RecordMessageConverter messageConverter) {
 		this.target.setMessageConverter(messageConverter);
 		return this;
 	}
 
+	/**
+	 * Set the payload type.
+	 * Only applies if a type-aware message converter is provided.
+	 * @param type the type to convert to.
+	 * @return the spec.
+	 */
 	public KafkaInboundChannelAdapterSpec<K, V> payloadType(Class<?> type) {
 		this.target.setPayloadType(type);
 		return this;
 	}
 
+	/**
+	 * Set a rebalance listener.
+	 * @param rebalanceListener the rebalance listener.
+	 * @return the spec.
+	 * @see ConsumerProperties
+	 * @deprecated in favor of using {@link ConsumerProperties}
+	 */
+	@Deprecated
 	public KafkaInboundChannelAdapterSpec<K, V> rebalanceListener(ConsumerRebalanceListener rebalanceListener) {
 		this.target.setRebalanceListener(rebalanceListener);
 		return this;
 	}
 
+	/**
+	 * Set to true to include the raw {@link ConsumerRecord} as headers with keys
+	 * {@link KafkaHeaders#RAW_DATA} and
+	 * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. enabling callers to have
+	 * access to the record to process errors.
+	 * @param rawMessageHeader true to include the header.
+	 * @return the spec.
+	 */
 	public KafkaInboundChannelAdapterSpec<K, V> rawMessageHeader(boolean rawMessageHeader) {
 		this.target.setRawMessageHeader(rawMessageHeader);
 		return this;