diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java index 2b75380436..1bf4ddf505 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java @@ -30,6 +30,7 @@ import org.springframework.util.StringUtils; * Parser for the inbound channel adapter. * * @author Gary Russell + * @author Anshul Mehra * @since 3.2 * */ @@ -44,6 +45,10 @@ public class KafkaInboundChannelAdapterParser extends AbstractPollingInboundChan 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 @@ public class KafkaInboundChannelAdapterParser extends AbstractPollingInboundChan 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(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java index 5dd14bbaa3..a8c5b806ae 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java @@ -27,6 +27,7 @@ import org.springframework.kafka.core.ConsumerFactory; 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 @@ import org.springframework.kafka.support.TopicPartitionOffset; * @author Artem Bilan * @author Nasko Vasilev * @author Gary Russell + * @author Anshul Mehra * * @since 3.0 */ @@ -79,28 +81,13 @@ public final class Kafka { * @param the Kafka message value type. * @return the spec. * @since 3.0.1 + * @deprecated in favor of {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties)} */ + @Deprecated public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( ConsumerFactory consumerFactory, String... topics) { - return inboundChannelAdapter(consumerFactory, false, topics); - } - - /** - * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and - * topics. - * @param consumerFactory the consumer factory. - * @param allowMultiFetch true to fetch multiple records on each poll. - * @param topics the topic(s). - * @param the Kafka message key type. - * @param the Kafka message value type. - * @return the spec. - * @since 3.2 - */ - public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( - ConsumerFactory consumerFactory, boolean allowMultiFetch, String... topics) { - - return new KafkaInboundChannelAdapterSpec<>(consumerFactory, allowMultiFetch, topics); + return inboundChannelAdapter(consumerFactory, new ConsumerProperties(topics), false); } /** @@ -113,7 +100,10 @@ public final class Kafka { * @param the Kafka message value type. * @return the spec. * @since 3.0.1 + * @deprecated in favor of + * {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)} */ + @Deprecated public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( ConsumerFactory consumerFactory, KafkaAckCallbackFactory ackCallbackFactory, String... topics) { @@ -132,14 +122,92 @@ public final class Kafka { * @param 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 KafkaInboundChannelAdapterSpec inboundChannelAdapter( ConsumerFactory consumerFactory, KafkaAckCallbackFactory ackCallbackFactory, boolean allowMultiFetch, String... topics) { - return new KafkaInboundChannelAdapterSpec<>(consumerFactory, ackCallbackFactory, allowMultiFetch, 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 the Kafka message key type. + * @param the Kafka message value type. + * @return the spec. + * @since 3.2 + */ + public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( + ConsumerFactory 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 the Kafka message key type. + * @param the Kafka message value type. + * @return the spec. + * @since 3.2 + */ + public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( + ConsumerFactory 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 the Kafka message key type. + * @param the Kafka message value type. + * @return the spec. + * @since 3.2 + */ + public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( + ConsumerFactory consumerFactory, + ConsumerProperties consumerProperties, + KafkaAckCallbackFactory 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 the Kafka message key type. + * @param the Kafka message value type. + * @return the spec. + * @since 3.2 + */ + public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( + ConsumerFactory consumerFactory, + ConsumerProperties consumerProperties, + KafkaAckCallbackFactory ackCallbackFactory, + boolean allowMultiFetch) { + + return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, ackCallbackFactory, allowMultiFetch); } /** diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java index a90e1db228..e5e879f89c 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java +++ b/spring-integration-kafka/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,6 +36,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter; * @param the value type. * * @author Gary Russell + * @author Anshul Mehra * * @since 3.0.1 * @@ -38,46 +44,147 @@ import org.springframework.kafka.support.converter.RecordMessageConverter; public class KafkaInboundChannelAdapterSpec extends MessageSourceSpec, KafkaMessageSource> { + /** + * 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 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 consumerFactory, KafkaAckCallbackFactory 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 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 consumerFactory, + ConsumerProperties consumerProperties, + KafkaAckCallbackFactory 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 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 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 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 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 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 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 rawMessageHeader(boolean rawMessageHeader) { this.target.setRawMessageHeader(rawMessageHeader); return this; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index 7c350bc7b5..361c1f027b 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -52,6 +52,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.listener.ConsumerAwareRebalanceListener; +import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.KafkaMessageHeaders; @@ -60,6 +61,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * Polled message source for kafka. Only one thread can poll for data (or @@ -90,8 +92,6 @@ import org.springframework.util.Assert; */ public class KafkaMessageSource extends AbstractMessageSource implements Pausable { - private static final long DEFAULT_POLL_TIMEOUT = 50L; - private static final long MIN_ASSIGN_TIMEOUT = 2000L; /** @@ -119,7 +119,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl private String clientId = "message.source"; - private Duration pollTimeout = Duration.ofMillis(DEFAULT_POLL_TIMEOUT); + private Duration pollTimeout; private RecordMessageConverter messageConverter = new MessagingMessageConverter(); @@ -135,7 +135,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl private boolean running; - private Duration assignTimeout = this.minTimeoutProvider.get(); + private Duration assignTimeout; private volatile Consumer consumer; @@ -150,48 +150,80 @@ public class KafkaMessageSource extends AbstractMessageSource impl /** * Construct an instance with the supplied parameters. Fetching multiple * records per poll will be disabled. - * * @param consumerFactory the consumer factory. * @param topics the topics. - * @see #KafkaMessageSource(ConsumerFactory, KafkaAckCallbackFactory, boolean, String...) + * @see #KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean) + * @deprecated in favor of {@link #KafkaMessageSource(ConsumerFactory, ConsumerProperties)} */ + @Deprecated public KafkaMessageSource(ConsumerFactory consumerFactory, String... topics) { - this(consumerFactory, new KafkaAckCallbackFactory<>(), false, topics); - } - - /** - * Construct an instance with the supplied parameters. Set 'allowMultiFetch' to true - * to allow up to {@code max.poll.records} to be fetched on each poll. When false - * (default) {@code max.poll.records} is coerced to 1 if the consumer factory is a - * {@link DefaultKafkaConsumerFactory} or otherwise rejected with an - * {@link IllegalArgumentException}. IMPORTANT: When true, you must call - * {@link #receive()} at a sufficient rate to consume the number of records received - * within {@code max.poll.interval.ms}. When false, you must call {@link #receive()} - * within {@code max.poll.interval.ms}. {@link #pause()} will not take effect until - * the records from the previous poll are consumed. - * - * @param consumerFactory the consumer factory. - * @param allowMultiFetch true to allow {@code max.poll.records > 1}. - * @param topics the topics. - * @since 3.2 - */ - public KafkaMessageSource(ConsumerFactory consumerFactory, boolean allowMultiFetch, String... topics) { - this(consumerFactory, new KafkaAckCallbackFactory<>(), allowMultiFetch, topics); + this(consumerFactory, new ConsumerProperties(topics), new KafkaAckCallbackFactory<>(), false); } /** * Construct an instance with the supplied parameters. Fetching multiple * records per poll will be disabled. - * * @param consumerFactory the consumer factory. * @param ackCallbackFactory the ack callback factory. * @param topics the topics. - * @see #KafkaMessageSource(ConsumerFactory, KafkaAckCallbackFactory, boolean, String...) + * @see #KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean) + * @deprecated in favor of + * {@link #KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)} */ + @Deprecated public KafkaMessageSource(ConsumerFactory consumerFactory, KafkaAckCallbackFactory ackCallbackFactory, String... topics) { - this(consumerFactory, ackCallbackFactory, false, topics); + this(consumerFactory, new ConsumerProperties(topics), ackCallbackFactory, false); + } + + /** + * Construct an instance with the supplied parameters. Fetching multiple + * records per poll will be disabled. + * @param consumerFactory the consumer factory. + * @param consumerProperties the consumer properties. + * @since 3.2 + * @see #KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean) + */ + public KafkaMessageSource(ConsumerFactory consumerFactory, ConsumerProperties consumerProperties) { + this(consumerFactory, consumerProperties, new KafkaAckCallbackFactory<>(), false); + } + + /** + * Construct an instance with the supplied parameters. Set 'allowMultiFetch' to true + * to allow up to {@code max.poll.records} to be fetched on each poll. When false + * (default) {@code max.poll.records} is coerced to 1 if the consumer factory is a + * {@link DefaultKafkaConsumerFactory} or otherwise rejected with an + * {@link IllegalArgumentException}. IMPORTANT: When true, you must call + * {@link #receive()} at a sufficient rate to consume the number of records received + * within {@code max.poll.interval.ms}. When false, you must call {@link #receive()} + * within {@code max.poll.interval.ms}. {@link #pause()} will not take effect until + * the records from the previous poll are consumed. + * @param consumerFactory the consumer factory. + * @param consumerProperties the consumer properties. + * @param allowMultiFetch true to allow {@code max.poll.records > 1}. + * @since 3.2 + */ + public KafkaMessageSource(ConsumerFactory consumerFactory, + ConsumerProperties consumerProperties, + boolean allowMultiFetch) { + this(consumerFactory, consumerProperties, new KafkaAckCallbackFactory<>(), allowMultiFetch); + } + + /** + * Construct an instance with the supplied parameters. Fetching multiple + * records per poll will be disabled. + * @param consumerFactory the consumer factory. + * @param consumerProperties the consumer properties. + * @param ackCallbackFactory the ack callback factory. + * @since 3.2 + * @see #KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean) + */ + public KafkaMessageSource(ConsumerFactory consumerFactory, + ConsumerProperties consumerProperties, + KafkaAckCallbackFactory ackCallbackFactory) { + + this(consumerFactory, consumerProperties, ackCallbackFactory, false); } /** @@ -206,20 +238,36 @@ public class KafkaMessageSource extends AbstractMessageSource impl * the records from the previous poll are consumed. * * @param consumerFactory the consumer factory. + * @param consumerProperties the consumer properties. * @param ackCallbackFactory the ack callback factory. * @param allowMultiFetch true to allow {@code max.poll.records > 1}. - * @param topics the topics. * @since 3.2 */ public KafkaMessageSource(ConsumerFactory consumerFactory, - KafkaAckCallbackFactory ackCallbackFactory, boolean allowMultiFetch, String... topics) { + ConsumerProperties consumerProperties, + KafkaAckCallbackFactory ackCallbackFactory, + boolean allowMultiFetch) { Assert.notNull(consumerFactory, "'consumerFactory' must not be null"); Assert.notNull(ackCallbackFactory, "'ackCallbackFactory' must not be null"); - Assert.isTrue(topics != null && topics.length > 0, "At least one topic is required"); + Assert.isTrue(consumerProperties.getTopics() != null && consumerProperties.getTopics().length > 0, "At least one topic is required"); this.consumerFactory = fixOrRejectConsumerFactory(consumerFactory, allowMultiFetch); this.ackCallbackFactory = ackCallbackFactory; - this.topics = topics; + this.topics = consumerProperties.getTopics(); + this.groupId = consumerProperties.getGroupId(); + if (StringUtils.hasText(consumerProperties.getClientId())) { + this.clientId = consumerProperties.getClientId(); + } + this.pollTimeout = Duration.ofMillis(consumerProperties.getPollTimeout()); + this.assignTimeout = this.minTimeoutProvider.get(); + this.commitTimeout = consumerProperties.getSyncCommitTimeout(); + this.ackCallbackFactory.setCommitTimeout(consumerProperties.getSyncCommitTimeout()); + if (consumerProperties.getConsumerRebalanceListener() instanceof ConsumerAwareRebalanceListener) { + this.consumerAwareRebalanceListener = (ConsumerAwareRebalanceListener) consumerProperties.getConsumerRebalanceListener(); + } + else { + this.rebalanceListener = consumerProperties.getConsumerRebalanceListener(); + } } protected String getGroupId() { @@ -229,7 +277,10 @@ public class KafkaMessageSource extends AbstractMessageSource impl /** * Set the group.id property for the consumer. * @param groupId the group id. + * @see ConsumerProperties + * @deprecated in favor of using {@link ConsumerProperties} */ + @Deprecated public void setGroupId(String groupId) { this.groupId = groupId; } @@ -241,7 +292,10 @@ public class KafkaMessageSource extends AbstractMessageSource impl /** * Set the client.id property for the consumer. * @param clientId the client id. + * @see ConsumerProperties + * @deprecated in favor of using {@link ConsumerProperties} */ + @Deprecated public void setClientId(String clientId) { this.clientId = clientId; } @@ -251,9 +305,12 @@ public class KafkaMessageSource extends AbstractMessageSource impl } /** - * Set the pollTimeout for the poll() operations; default 50ms. + * Set the pollTimeout for the poll() operations. * @param pollTimeout the poll timeout. + * @see ConsumerProperties + * @deprecated in favor of using {@link ConsumerProperties} */ + @Deprecated public void setPollTimeout(long pollTimeout) { this.pollTimeout = Duration.ofMillis(pollTimeout); this.assignTimeout = this.minTimeoutProvider.get(); @@ -292,7 +349,10 @@ public class KafkaMessageSource extends AbstractMessageSource impl /** * Set a rebalance listener. * @param rebalanceListener the rebalance listener. + * @see ConsumerProperties + * @deprecated in favor of using {@link ConsumerProperties} */ + @Deprecated public void setRebalanceListener(ConsumerRebalanceListener rebalanceListener) { this.rebalanceListener = rebalanceListener; if (rebalanceListener instanceof ConsumerAwareRebalanceListener) { @@ -324,16 +384,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl return this.commitTimeout; } - /** - * Set the timeout for commits. - * @param commitTimeout the timeout. - * @since 3.2 - */ - public void setCommitTimeout(Duration commitTimeout) { - this.commitTimeout = commitTimeout; - this.ackCallbackFactory.setCommitTimeout(commitTimeout); - } - private ConsumerFactory fixOrRejectConsumerFactory(ConsumerFactory suppliedConsumerFactory, boolean allowMultiFetch) { diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd index 16c25fa994..7b6080ac2b 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd @@ -156,6 +156,18 @@ + + + + A reference to a ConsumerProperties. + + + + + + + + @@ -172,28 +184,36 @@ + [DEPRECATED] Comma-delimited list of topic names. + Deprecated in favor of 'consumer-properties'. - + + [DEPRECATED] The 'group.id' Kafka Consumer property; overrides the value in the consumer factory. + Deprecated in favor of 'consumer-properties'. + [DEPRECATED] The 'client.id' Kafka Consumer property; overrides the value in the consumer factory. + Deprecated in favor of 'consumer-properties'. - The time to block in poll() in milliseconds; default 50ms. + [DEPRECATED] + The time to block in poll() in milliseconds; default 5000ms. + Deprecated in favor of 'consumer-properties'. @@ -225,12 +245,14 @@ + [DEPRECATED] A reference to a 'ConsumerRebalanceListener'. + Deprecated in favor of 'consumer-properties'. + type="org.apache.kafka.clients.consumer.ConsumerRebalanceListener"/> diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml index 779d4d62c3..9cec89bea7 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml @@ -29,11 +29,15 @@ + + + + + + consumer-properties="cp"> diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml index ce706053a2..df661cb319 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml @@ -10,25 +10,21 @@ + auto-startup="false"> @@ -42,6 +38,18 @@ + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 291a456250..0ece863e7a 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -57,6 +57,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.ContainerProperties.AckMode; import org.springframework.kafka.listener.GenericMessageListenerContainer; @@ -83,6 +84,7 @@ import org.springframework.test.context.junit4.SpringRunner; * @author Nasko Vasilev * @author Biju Kunjummen * @author Gary Russell + * @author Anshul Mehra * * @since 3.0 */ @@ -337,7 +339,7 @@ public class KafkaDslTests { @Bean public IntegrationFlow sourceFlow() { return IntegrationFlows - .from(Kafka.inboundChannelAdapter(consumerFactory(), TEST_TOPIC3), + .from(Kafka.inboundChannelAdapter(consumerFactory(), new ConsumerProperties(TEST_TOPIC3)), e -> e.poller(Pollers.fixedDelay(100))) .handle(p -> { this.fromSource = p.getPayload(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java index ffec00fcf0..5459a37196 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java @@ -32,6 +32,7 @@ import org.junit.Test; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -40,6 +41,7 @@ import org.springframework.messaging.Message; /** * @author Gary Russell * @author Artem Bilan + * @author Anshul Mehra * * @since 3.0.1 * @@ -59,9 +61,9 @@ public class MessageSourceIntegrationTests { consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps); - KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, TOPIC1); + ConsumerProperties consumerProperties = new ConsumerProperties(TOPIC1); final CountDownLatch assigned = new CountDownLatch(1); - source.setRebalanceListener(new ConsumerRebalanceListener() { + consumerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection partitions) { @@ -73,6 +75,7 @@ public class MessageSourceIntegrationTests { } }); + KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, consumerProperties); Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(producerProps); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index dbaa04a252..7793dda65c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -62,6 +62,7 @@ import org.springframework.integration.acks.AcknowledgmentCallback; import org.springframework.integration.test.util.TestUtils; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; @@ -111,7 +112,7 @@ public class MessageSourceTests { willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory) .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); - KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo")); source.setRawMessageHeader(true); Message received = source.receive(); @@ -205,7 +206,7 @@ public class MessageSourceTests { willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory) .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); - KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo")); Message received1 = source.receive(); consumer.paused(); // need some other interaction with mock between polls for InOrder @@ -279,8 +280,9 @@ public class MessageSourceTests { willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory) .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); - KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); - source.setCommitTimeout(Duration.ofSeconds(30)); + ConsumerProperties consumerProperties = new ConsumerProperties("foo"); + consumerProperties.setSyncCommitTimeout(Duration.ofSeconds(30)); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, consumerProperties); Message received = source.receive(); assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L); @@ -347,7 +349,7 @@ public class MessageSourceTests { willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory) .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); - KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo")); Message received1 = source.receive(); consumer.paused(); // need some other interaction with mock between polls for InOrder @@ -407,17 +409,17 @@ public class MessageSourceTests { @Test public void testMaxPollRecords() { KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap()), - "topic"); + new ConsumerProperties("topic")); assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class) .get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1); source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>( - Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)), "topic"); + Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)), new ConsumerProperties("topic")); assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class) .get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1); try { new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) { - }), "topic"); + }), new ConsumerProperties("topic")); fail("Expected exception"); } catch (IllegalArgumentException e) { @@ -441,17 +443,17 @@ public class MessageSourceTests { records1.put(topicPartition, Arrays.asList( new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); ConsumerRecords cr1 = new ConsumerRecords(records1); - given(consumer.poll(Duration.of(2, ChronoUnit.SECONDS))).willReturn(cr1, ConsumerRecords.EMPTY); + given(consumer.poll(Duration.of(20 * 5000, ChronoUnit.MILLIS))).willReturn(cr1, ConsumerRecords.EMPTY); Map> records2 = new LinkedHashMap<>(); records2.put(topicPartition, Arrays.asList( new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); ConsumerRecords cr2 = new ConsumerRecords(records2); - given(consumer.poll(Duration.of(50, ChronoUnit.MILLIS))).willReturn(cr2, ConsumerRecords.EMPTY); + given(consumer.poll(Duration.of(5000, ChronoUnit.MILLIS))).willReturn(cr2, ConsumerRecords.EMPTY); ConsumerFactory consumerFactory = mock(ConsumerFactory.class); willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory) .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); - KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo")); source.setRawMessageHeader(true); Message received = source.receive(); @@ -477,13 +479,13 @@ public class MessageSourceTests { InOrder inOrder = inOrder(consumer); inOrder.verify(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); // assignTimeout used on initial poll (before partition assigned) - inOrder.verify(consumer).poll(Duration.of(2, ChronoUnit.SECONDS)); + inOrder.verify(consumer).poll(Duration.of(20 * 5000, ChronoUnit.MILLIS)); inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L))); // pollTimeout used on subsequent polls - inOrder.verify(consumer).poll(Duration.of(50, ChronoUnit.MILLIS)); + inOrder.verify(consumer).poll(Duration.of(5000, ChronoUnit.MILLIS)); inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L))); // assignTimeout used after partitions revoked - inOrder.verify(consumer).poll(Duration.of(2, ChronoUnit.SECONDS)); + inOrder.verify(consumer).poll(Duration.of(20 * 5000, ChronoUnit.MILLIS)); } @SuppressWarnings({ "rawtypes", "unchecked" }) @@ -513,7 +515,7 @@ public class MessageSourceTests { willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 4)).given(consumerFactory) .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); - KafkaMessageSource source = new KafkaMessageSource(consumerFactory, true, "foo"); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, new ConsumerProperties("foo"), true); source.setRawMessageHeader(true); Message received = source.receive(); diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt index 017073cb95..0ff59bac09 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt @@ -17,15 +17,12 @@ package org.springframework.integration.kafka.dsl import assertk.assertThat -import assertk.assertions.contains import assertk.assertions.isEqualTo import assertk.assertions.isInstanceOf import assertk.assertions.isNotNull import assertk.assertions.isNull import assertk.assertions.isSameAs import assertk.assertions.isTrue -import assertk.catch -import kafka.tools.ConsoleProducer import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.clients.consumer.ConsumerRebalanceListener import org.apache.kafka.clients.producer.ProducerConfig @@ -44,7 +41,6 @@ import org.springframework.integration.config.EnableIntegration import org.springframework.integration.dsl.IntegrationFlow import org.springframework.integration.dsl.IntegrationFlows import org.springframework.integration.dsl.Pollers -import org.springframework.integration.expression.ValueExpression import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler @@ -57,6 +53,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory import org.springframework.kafka.core.DefaultKafkaProducerFactory import org.springframework.kafka.core.KafkaTemplate import org.springframework.kafka.core.ProducerFactory +import org.springframework.kafka.listener.ConsumerProperties import org.springframework.kafka.listener.ContainerProperties import org.springframework.kafka.listener.GenericMessageListenerContainer import org.springframework.kafka.listener.KafkaMessageListenerContainer @@ -320,7 +317,7 @@ class KafkaDslKotlinTests { @Bean fun sourceFlow() = IntegrationFlows - .from(Kafka.inboundChannelAdapter(consumerFactory(), TEST_TOPIC3)) { e -> e.poller(Pollers.fixedDelay(100)) } + .from(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3))) { e -> e.poller(Pollers.fixedDelay(100)) } .handle({ p -> this.fromSource = p.getPayload() this.sourceFlowLatch.countDown()