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

Use ConsumerProperties in KafkaMessageSource

Browse Source 
* Preserve existing constructors

* Add @deprecated in java docs
This commit is contained in:
Anshul Mehra
2019-08-23 14:07:44 -04:00
committed by Artem Bilan
parent e6cdb8818f
commit e52c535f7c
11 changed files with 375 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java
View File

@@ -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();
}

106
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java
View File

@@ -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 <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);
}
/**
* 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 <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, 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 <V> the Kafka message value type.
* @return the spec.
* @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,
KafkaAckCallbackFactory<K, V> ackCallbackFactory, String... topics) {
@@ -132,14 +122,92 @@ public final class Kafka {
* @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,
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 <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 <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,
boolean allowMultiFetch) {
return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, ackCallbackFactory, allowMultiFetch);
}
/**

111
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java
View File

@@ -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 <V> 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<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;

140
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
View File

@@ -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<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> impl
private boolean running;
private Duration assignTimeout = this.minTimeoutProvider.get();
private Duration assignTimeout;
private volatile Consumer<K, V> consumer;
@@ -150,48 +150,80 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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<K, V> 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<K, V> 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<K, V> consumerFactory,
KafkaAckCallbackFactory<K, V> 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<K, V> 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<K, V> 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<K, V> consumerFactory,
ConsumerProperties consumerProperties,
KafkaAckCallbackFactory<K, V> ackCallbackFactory) {
this(consumerFactory, consumerProperties, ackCallbackFactory, false);
}
/**
@@ -206,20 +238,36 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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<K, V> consumerFactory,
KafkaAckCallbackFactory<K, V> ackCallbackFactory, boolean allowMultiFetch, String... topics) {
ConsumerProperties consumerProperties,
KafkaAckCallbackFactory<K, V> 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<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> 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<K, V> fixOrRejectConsumerFactory(ConsumerFactory<K, V> suppliedConsumerFactory,
boolean allowMultiFetch) {

28
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd
View File

@@ -156,6 +156,18 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="consumer-properties" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A reference to a ConsumerProperties.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.kafka.listener.ConsumerProperties"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="ack-factory" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -172,28 +184,36 @@
<xsd:attribute name="topics" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
[DEPRECATED]
Comma-delimited list of topic names.
Deprecated in favor of 'consumer-properties'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="group-id" use="required" type="xsd:string">
<xsd:attribute name="group-id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
[DEPRECATED]
The 'group.id' Kafka Consumer property; overrides the value in the consumer factory.
Deprecated in favor of 'consumer-properties'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="client-id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
[DEPRECATED]
The 'client.id' Kafka Consumer property; overrides the value in the consumer factory.
Deprecated in favor of 'consumer-properties'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="poll-timeout">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
@@ -225,12 +245,14 @@
<xsd:attribute name="rebalance-listener" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
[DEPRECATED]
A reference to a 'ConsumerRebalanceListener'.
Deprecated in favor of 'consumer-properties'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.apache.kafka.clients.consumer.ConsumerRebalanceListener"/>
type="org.apache.kafka.clients.consumer.ConsumerRebalanceListener"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>

8
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml
View File

@@ -29,11 +29,15 @@
</constructor-arg>
</bean>
<bean id="cp" class="org.springframework.kafka.listener.ConsumerProperties">
<constructor-arg name="topics" value="one"/>
<property name="groupId" value="groupForTopic1"/>
</bean>
<int-kafka:inbound-channel-adapter
channel="fromOne"
consumer-factory="cf"
topics="one"
group-id="groupForTopic1">
consumer-properties="cp">
<int:poller fixed-delay="5000"/>
</int-kafka:inbound-channel-adapter>

22
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml
View File

@@ -10,25 +10,21 @@
<int-kafka:inbound-channel-adapter
id="adapter1"
consumer-factory="consumerFactory"
consumer-properties="consumerProperties1"
ack-factory="ackFactory"
topics="topic1"
channel="inbound"
client-id="client"
group-id="group"
message-converter="converter"
payload-type="java.lang.String"
raw-header="true"
auto-startup="false"
rebalance-listener="rebal">
auto-startup="false">
<int:poller fixed-delay="5000"/>
</int-kafka:inbound-channel-adapter>
<int-kafka:inbound-channel-adapter
id="adapter2"
consumer-factory="multiFetchConsumerFactory"
consumer-properties="consumerProperties2"
allow-multi-fetch="true"
topics="topic1, topic2"
group-id="group"
auto-startup="false"
channel="inbound">
<int:poller fixed-delay="5000"/>
@@ -42,6 +38,18 @@
</constructor-arg>
</bean>
<bean id="consumerProperties1" class="org.springframework.kafka.listener.ConsumerProperties">
<constructor-arg name="topics" value="topic1"/>
<property name="groupId" value="group"/>
<property name="clientId" value="client"/>
<property name="consumerRebalanceListener" ref="rebal"/>
</bean>
<bean id="consumerProperties2" class="org.springframework.kafka.listener.ConsumerProperties">
<constructor-arg name="topics" value="topic1, topic2"/>
<property name="groupId" value="group"/>
</bean>
<bean id="multiFetchConsumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<map>

4
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -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();

7
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java
View File

@@ -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<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
KafkaMessageSource<Integer, String> 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<TopicPartition> partitions) {
@@ -73,6 +75,7 @@ public class MessageSourceIntegrationTests {
}
});
KafkaMessageSource<Integer, String> source = new KafkaMessageSource<>(consumerFactory, consumerProperties);
Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Object, Object> producerFactory = new DefaultKafkaProducerFactory<>(producerProps);

32
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java
View File

@@ -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<TopicPartition, List<ConsumerRecord>> 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();

7
spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt
View File

@@ -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()