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 cc09688c1..ffcb76f23 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 @@ -162,6 +162,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder { private static final int DEFAULT_ZK_CONNECTION_TIMEOUT = 10000; + private static final boolean DEFAULT_SYNC_PRODUCER = false; + private static final StartOffset DEFAULT_START_OFFSET = StartOffset.latest; private RetryOperations retryOperations; @@ -256,6 +258,8 @@ public class KafkaMessageChannelBinder extends AbstractBinder { private int zkConnectionTimeout = DEFAULT_ZK_CONNECTION_TIMEOUT; + private boolean syncProducer = DEFAULT_SYNC_PRODUCER; + private ProducerListener producerListener; public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress, @@ -296,6 +300,14 @@ public class KafkaMessageChannelBinder extends AbstractBinder { this.offsetUpdateShutdownTimeout = offsetUpdateShutdownTimeout; } + public boolean isSyncProducer() { + return this.syncProducer; + } + + public void setSyncProducer(boolean syncProducer) { + this.syncProducer = syncProducer; + } + public ConnectionFactory getConnectionFactory() { return connectionFactory; } @@ -470,6 +482,7 @@ public class KafkaMessageChannelBinder extends AbstractBinder { ProducerMetadata producerMetadata = new ProducerMetadata<>( name, byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER); + producerMetadata.setSync(isSyncProducer()); producerMetadata.setCompressionType(ProducerMetadata.CompressionType.valueOf( producerPropertiesAccessor.getCompressionCodec(this.defaultCompressionCodec))); producerMetadata.setBatchBytes(producerPropertiesAccessor.getBatchSize(this.defaultBatchSize)); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java index a03fc1cde..1be2b5cf9 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java @@ -61,6 +61,11 @@ class KafkaBinderConfigurationProperties { */ private int zkConnectionTimeout; + /** + * Flag to indicate if the Kafka Producer is synchronous or asynchronous. + */ + private boolean syncProducer = false; + public String getZkConnectionString() { return toConnectionString(this.zkNodes, this.defaultZkPort); } @@ -142,6 +147,7 @@ class KafkaBinderConfigurationProperties { this.resetOffsets = resetOffsets; } + public int getZkSessionTimeout() { return this.zkSessionTimeout; } @@ -158,6 +164,14 @@ class KafkaBinderConfigurationProperties { this.zkConnectionTimeout = zkConnectionTimeout; } + public boolean isSyncProducer() { + return this.syncProducer; + } + + public void setSyncProducer(boolean syncProducer) { + this.syncProducer = syncProducer; + } + /** * Converts an array of host values to a comma-separated String. * 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 84bd9c08e..9c8a2c5d3 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 @@ -88,6 +88,8 @@ public class KafkaServiceAutoConfiguration { kafkaMessageChannelBinder.setZkSessionTimeout(kafkaBinderConfigurationProperties.getZkSessionTimeout()); kafkaMessageChannelBinder.setZkConnectionTimeout(kafkaBinderConfigurationProperties.getZkConnectionTimeout()); + kafkaMessageChannelBinder.setSyncProducer(kafkaBinderConfigurationProperties.isSyncProducer()); + kafkaMessageChannelBinder.setDefaultAutoCommitEnabled(kafkaBinderDefaultProperties.isAutoCommitEnabled()); kafkaMessageChannelBinder.setDefaultBatchSize(kafkaBinderDefaultProperties.getBatchSize()); kafkaMessageChannelBinder.setDefaultBatchTimeout(kafkaBinderDefaultProperties.getBatchTimeout()); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties index ef3773e0f..e786029c1 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties @@ -8,6 +8,7 @@ spring.cloud.stream.binder.kafka.offsetUpdateCount=0 spring.cloud.stream.binder.kafka.offsetUpdateShutdownTimeout=2000 spring.cloud.stream.binder.kafka.zkSessionTimeout=10000 spring.cloud.stream.binder.kafka.zkConnectionTimeout=10000 +spring.cloud.stream.binder.kafka.syncProducer=false spring.cloud.stream.binder.kafka.default.batchSize=16384 spring.cloud.stream.binder.kafka.default.batchTimeout=0 spring.cloud.stream.binder.kafka.default.requiredAcks=1 diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 5e3587352..be57211eb 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -23,6 +23,7 @@ import static org.hamcrest.collection.IsCollectionWithSize.hasSize; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.util.Arrays; import java.util.Collection; @@ -36,6 +37,7 @@ import org.junit.Before; import org.junit.ClassRule; import org.junit.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderPropertyKeys; import org.springframework.cloud.stream.binder.Binding; @@ -49,9 +51,11 @@ import org.springframework.integration.kafka.core.KafkaMessage; import org.springframework.integration.kafka.core.Partition; import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; import org.springframework.integration.kafka.listener.MessageListener; +import org.springframework.integration.kafka.support.ProducerConfiguration; import org.springframework.integration.kafka.support.ZookeeperConnect; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; import org.springframework.messaging.support.GenericMessage; import kafka.api.OffsetRequest; @@ -481,4 +485,26 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { consumerBinding.unbind(); producerBinding.unbind(); } + + @Test + public void testSyncProducerMetadata() throws Exception { + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), + kafkaTestSupport.getBrokerAddress(), kafkaTestSupport.getZkConnectString()); + binder.setSyncProducer(true); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + + DirectChannel output = new DirectChannel(); + + String testTopicName = UUID.randomUUID().toString(); + Binding producerBinding = binder.bindProducer(testTopicName, output, null); + DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding)); + MessageHandler handler = (MessageHandler) accessor.getPropertyValue("handler"); + DirectFieldAccessor accessor1 = new DirectFieldAccessor(handler); + ProducerConfiguration producerConfiguration = (ProducerConfiguration) accessor1.getPropertyValue("producerConfiguration"); + assertTrue("Kafka Sync Producer should have been enabled.", producerConfiguration.getProducerMetadata().isSync()); + producerBinding.unbind(); + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 757036f16..4c50628dd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -85,7 +85,6 @@ public class BindingProperties { private Integer batchTimeout; // Inbound properties - private Integer concurrency; // Partition properties @@ -284,11 +283,11 @@ public class BindingProperties { sb.append("partitionSelectorClass=" + partitionSelectorClass); sb.append(COMMA); } - if (this.partitionSelectorClass != null && !this.partitionSelectorClass.isEmpty()) { + if (this.partitionSelectorExpression != null && !this.partitionSelectorExpression.isEmpty()) { sb.append("partitionSelectorExpression=" + partitionSelectorExpression); sb.append(COMMA); } - if (this.partitioned) { + if (this.partitionCount != null) { sb.append("partitionCount=" + this.partitionCount); sb.append(COMMA); } @@ -316,8 +315,10 @@ public class BindingProperties { sb.append("batchTimeout=" + this.batchTimeout); sb.append(COMMA); } - sb.append("partitioned=" + this.partitioned); - sb.append(COMMA); + if (this.partitioned != null) { + sb.append("partitioned=" + this.partitioned); + sb.append(COMMA); + } if (this.partitionIndex != null) { sb.append("partitionIndex=" + this.partitionIndex); sb.append(COMMA);