Support sync producer option in Kafka binder

- Add a binding property `spring.cloud.stream.bindings.<outputChannelName>.syncProducer`
   When set to `true`, the Kafka producer metadata is set to use `sync` producer
 - Update DefaultPropertiesAccessor for Kafka to use the `syncProducer` property
 - Add test

This resolves #343

Move syncProducer as a binder level property

 - Add a property `syncProducer` in Kafka Binder configuration properties (prefix: spring.cloud.stream.binder.kafka)
  - Set Kafka ProducerMeatadata `sync` based on the value set in this property
 - Update test

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<MessageChannel> {
 
 	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<MessageChannel> {
 
 	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<MessageChannel> {
 		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<MessageChannel> {
 
 		ProducerMetadata<byte[], byte[]> 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));

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.
 	 *

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

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

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

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