diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml index ae278d37e..1601c41d8 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml @@ -14,8 +14,8 @@ - 0.8.2.1 - 1.2.0.RELEASE + 0.8.2.2 + 1.3.0.RELEASE 1.0.0 diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index e771c13cd..5c4198518 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -55,15 +55,8 @@ import org.springframework.integration.kafka.core.DefaultConnectionFactory; import org.springframework.integration.kafka.core.Partition; import org.springframework.integration.kafka.core.ZookeeperConfiguration; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; -import org.springframework.integration.kafka.listener.Acknowledgment; -import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; -import org.springframework.integration.kafka.listener.KafkaTopicOffsetManager; -import org.springframework.integration.kafka.listener.OffsetManager; -import org.springframework.integration.kafka.support.KafkaHeaders; -import org.springframework.integration.kafka.support.ProducerConfiguration; -import org.springframework.integration.kafka.support.ProducerFactoryBean; -import org.springframework.integration.kafka.support.ProducerMetadata; -import org.springframework.integration.kafka.support.ZookeeperConnect; +import org.springframework.integration.kafka.listener.*; +import org.springframework.integration.kafka.support.*; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -124,6 +117,7 @@ import scala.collection.Seq; * @author David Turanski * @author Gary Russell * @author Mark Fisher + * @author Soby Chacko */ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { @@ -250,26 +244,12 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { private ConnectionFactory connectionFactory; - private String offsetStoreTopic = "SpringXdOffsets"; - // auto commit property private boolean defaultAutoCommitEnabled = DEFAULT_AUTO_COMMIT_ENABLED; private int socketBufferSize = 2097152; - private int offsetStoreSegmentSize = 250 * 1024 * 1024; - - private int offsetStoreRetentionTime = 60000; - - private int offsetStoreRequiredAcks = 1; - - private int offsetStoreMaxFetchSize = 1048576; - - private int offsetStoreBatchBytes = 200; - - private int offsetStoreBatchTime = 1000; - private int offsetUpdateTimeWindow = 10000; private int offsetUpdateCount = 0; @@ -282,8 +262,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { private StartOffset startOffset = DEFAULT_START_OFFSET; + private ProducerListener producerListener; + public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress, - String... headersToMap) { + String... headersToMap) { this.zookeeperConnect = zookeeperConnect; this.brokers = brokers; this.zkAddress = zkAddress; @@ -300,30 +282,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { } } - public void setOffsetStoreTopic(String offsetStoreTopic) { - this.offsetStoreTopic = offsetStoreTopic; - } - - public void setOffsetStoreSegmentSize(int offsetStoreSegmentSize) { - this.offsetStoreSegmentSize = offsetStoreSegmentSize; - } - - public void setOffsetStoreRetentionTime(int offsetStoreRetentionTime) { - this.offsetStoreRetentionTime = offsetStoreRetentionTime; - } - public void setSocketBufferSize(int socketBufferSize) { this.socketBufferSize = socketBufferSize; } - public void setOffsetStoreRequiredAcks(int offsetStoreRequiredAcks) { - this.offsetStoreRequiredAcks = offsetStoreRequiredAcks; - } - - public void setOffsetStoreMaxFetchSize(int offsetStoreMaxFetchSize) { - this.offsetStoreMaxFetchSize = offsetStoreMaxFetchSize; - } - public void setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow) { this.offsetUpdateTimeWindow = offsetUpdateTimeWindow; } @@ -336,18 +298,14 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { this.offsetUpdateShutdownTimeout = offsetUpdateShutdownTimeout; } - public void setOffsetStoreBatchBytes(int offsetStoreBatchBytes) { - this.offsetStoreBatchBytes = offsetStoreBatchBytes; - } - - public void setOffsetStoreBatchTime(int offsetStoreBatchTime) { - this.offsetStoreBatchTime = offsetStoreBatchTime; - } - public ConnectionFactory getConnectionFactory() { return connectionFactory; } + public void setProducerListener(ProducerListener producerListener) { + this.producerListener = producerListener; + } + /** * Retry configuration for operations such as validating topic creation * @param retryOperations the retry configuration @@ -504,6 +462,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { try { final ProducerConfiguration producerConfiguration = new ProducerConfiguration<>( producerMetadata, producerFB.getObject()); + producerConfiguration.setProducerListener(producerListener); MessageHandler handler = new SendingHandler(name, producerPropertiesAccessor, partitions.size(), producerConfiguration); @@ -715,18 +674,12 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { private OffsetManager createOffsetManager(String group, long referencePoint) { try { - KafkaTopicOffsetManager kafkaOffsetManager = - new KafkaTopicOffsetManager(zookeeperConnect, offsetStoreTopic, Collections. emptyMap()); kafkaOffsetManager.setConsumerId(group); kafkaOffsetManager.setReferenceTimestamp(referencePoint); - kafkaOffsetManager.setSegmentSize(offsetStoreSegmentSize); - kafkaOffsetManager.setRetentionTime(offsetStoreRetentionTime); - kafkaOffsetManager.setRequiredAcks(offsetStoreRequiredAcks); - kafkaOffsetManager.setMaxSize(offsetStoreMaxFetchSize); - kafkaOffsetManager.setBatchBytes(offsetStoreBatchBytes); - kafkaOffsetManager.setMaxQueueBufferingTime(offsetStoreBatchTime); - kafkaOffsetManager.afterPropertiesSet(); WindowingOffsetManager windowingOffsetManager = new WindowingOffsetManager(kafkaOffsetManager); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java index 00fe867a8..493b78e78 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.config; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration; @@ -25,6 +26,8 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.codec.Codec; +import org.springframework.integration.kafka.support.LoggingProducerListener; +import org.springframework.integration.kafka.support.ProducerListener; import org.springframework.integration.kafka.support.ZookeeperConnect; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -33,6 +36,7 @@ import org.springframework.util.StringUtils; * @author David Turanski * @author Marius Bogoevici * @author Mark Fisher + * @author Soby Chacko */ @Configuration @Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class}) @@ -51,20 +55,6 @@ public class KafkaMessageChannelBinderConfiguration { private KafkaMessageChannelBinder.Mode mode; - private String offsetStoreTopic; - - private int offsetStoreSegmentSize; - - private int offsetStoreRetentionTime; - - private int offsetStoreRequiredAcks; - - private int offsetStoreMaxFetchSize; - - private int offsetStoreBatchBytes; - - private int offsetStoreBatchTime; - private int offsetUpdateTimeWindow; private int offsetUpdateCount; @@ -81,6 +71,9 @@ public class KafkaMessageChannelBinderConfiguration { @Autowired private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties; + @Autowired + private ProducerListener producerListener; + @Bean ZookeeperConnect zookeeperConnect() { ZookeeperConnect zookeeperConnect = new ZookeeperConnect(); @@ -96,13 +89,6 @@ public class KafkaMessageChannelBinderConfiguration { headers); kafkaMessageChannelBinder.setCodec(codec); kafkaMessageChannelBinder.setMode(mode); - kafkaMessageChannelBinder.setOffsetStoreTopic(offsetStoreTopic); - kafkaMessageChannelBinder.setOffsetStoreSegmentSize(offsetStoreSegmentSize); - kafkaMessageChannelBinder.setOffsetStoreRetentionTime(offsetStoreRetentionTime); - kafkaMessageChannelBinder.setOffsetStoreRequiredAcks(offsetStoreRequiredAcks); - kafkaMessageChannelBinder.setOffsetStoreMaxFetchSize(offsetStoreMaxFetchSize); - kafkaMessageChannelBinder.setOffsetStoreBatchBytes(offsetStoreBatchBytes); - kafkaMessageChannelBinder.setOffsetStoreBatchTime(offsetStoreBatchTime); kafkaMessageChannelBinder.setOffsetUpdateTimeWindow(offsetUpdateTimeWindow); kafkaMessageChannelBinder.setOffsetUpdateCount(offsetUpdateCount); kafkaMessageChannelBinder.setOffsetUpdateShutdownTimeout(offsetUpdateShutdownTimeout); @@ -124,6 +110,8 @@ public class KafkaMessageChannelBinderConfiguration { kafkaMessageChannelBinder.setResetOffsets(resetOffsets); kafkaMessageChannelBinder.setStartOffset(startOffset); + kafkaMessageChannelBinder.setProducerListener(producerListener); + return kafkaMessageChannelBinder; } @@ -155,34 +143,6 @@ public class KafkaMessageChannelBinderConfiguration { this.mode = mode; } - public void setOffsetStoreTopic(String offsetStoreTopic) { - this.offsetStoreTopic = offsetStoreTopic; - } - - public void setOffsetStoreSegmentSize(int offsetStoreSegmentSize) { - this.offsetStoreSegmentSize = offsetStoreSegmentSize; - } - - public void setOffsetStoreRetentionTime(int offsetStoreRetentionTime) { - this.offsetStoreRetentionTime = offsetStoreRetentionTime; - } - - public void setOffsetStoreRequiredAcks(int offsetStoreRequiredAcks) { - this.offsetStoreRequiredAcks = offsetStoreRequiredAcks; - } - - public void setOffsetStoreMaxFetchSize(int offsetStoreMaxFetchSize) { - this.offsetStoreMaxFetchSize = offsetStoreMaxFetchSize; - } - - public void setOffsetStoreBatchBytes(int offsetStoreBatchBytes) { - this.offsetStoreBatchBytes = offsetStoreBatchBytes; - } - - public void setOffsetStoreBatchTime(int offsetStoreBatchTime) { - this.offsetStoreBatchTime = offsetStoreBatchTime; - } - public void setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow) { this.offsetUpdateTimeWindow = offsetUpdateTimeWindow; } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java index ec4079835..3318b106b 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java @@ -20,13 +20,17 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.autoconfigure.test.ImportAutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.Binder; +import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.PropertySource; +import org.springframework.integration.kafka.support.LoggingProducerListener; +import org.springframework.integration.kafka.support.ProducerListener; /** * Bind to Kafka services. * * @author Marius Bogoevici + * @author Soby Chacko */ @Configuration @ConditionalOnMissingBean(Binder.class) @@ -39,4 +43,10 @@ public class KafkaServiceAutoConfiguration { public static class DefaultProperties { } + + @Bean + @ConditionalOnMissingBean(ProducerListener.class) + ProducerListener producerListener() { + return new LoggingProducerListener(); + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index 5bea73141..544bedd0c 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -26,6 +26,8 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.codec.Codec; import org.springframework.integration.codec.kryo.KryoRegistrar; import org.springframework.integration.codec.kryo.PojoCodec; +import org.springframework.integration.kafka.support.LoggingProducerListener; +import org.springframework.integration.kafka.support.ProducerListener; import org.springframework.integration.kafka.support.ZookeeperConnect; import com.esotericsoftware.kryo.Kryo; @@ -39,6 +41,7 @@ import com.esotericsoftware.kryo.Registration; * @author Marius Bogoevici * @author David Turanski * @author Gary Russell + * @author Soby Chacko */ public class KafkaTestBinder extends AbstractTestBinder { @@ -59,6 +62,8 @@ public class KafkaTestBinder extends AbstractTestBinder