SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade Spring Integration Kafka to 1.3 in Kafka binder

Browse Source 
Upgrade scala/SIK versions in mvn pom

Migrage to KafkaNativeOffsetManager from KafkaTopicOffsetManager

Introduce LoggingProducerListener for errors - See GH #151

Remove properties specific to the Topic offset manager
This commit is contained in:
Soby Chacko
2016-02-05 18:02:47 -05:00
committed by Marius Bogoevici
parent ccc6049adc
commit 7138f125ee
5 changed files with 40 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml
View File

@@ -14,8 +14,8 @@
</parent>
<properties>
<kafka.version>0.8.2.1</kafka.version>
<spring-integration-kafka.version>1.2.0.RELEASE</spring-integration-kafka.version>
<kafka.version>0.8.2.2</kafka.version>
<spring-integration-kafka.version>1.3.0.RELEASE</spring-integration-kafka.version>
<rxjava-math.version>1.0.0</rxjava-math.version>
</properties>

75
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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<byte[], byte[]> 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.<Partition,
KafkaNativeOffsetManager kafkaOffsetManager =
new KafkaNativeOffsetManager(connectionFactory, zookeeperConnect,Collections.<Partition,
Long> 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);

58
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java
View File

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

10
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java
View File

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

5
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java
View File

@@ -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<KafkaMessageChannelBinder> {
@@ -59,6 +62,8 @@ public class KafkaTestBinder extends AbstractTestBinder<KafkaMessageChannelBinde
binder.setCodec(getCodec());
binder.setDefaultBatchingEnabled(false);
binder.setMode(mode);
ProducerListener producerListener = new LoggingProducerListener();
binder.setProducerListener(producerListener);
GenericApplicationContext context = new GenericApplicationContext();
context.refresh();
binder.setApplicationContext(context);