Support for sending messages synchronously to the Kafka broker.

Adds two fields to the producer-configuration that allows to set the send mode (async or sync) and an optional timeout in sync mode. Default behavior is async.
This allows for an upstream application to verify that the message have been handed off to Kafka following the delivery options specified, and not just residing in memory at the producer.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java

@@ -127,6 +127,11 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars
 					"compression-type");
 			IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
 					"batch-bytes");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
+					"sync");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(producerMetadataBuilder, producerConfiguration,
+					"send-timeout");
+
 			AbstractBeanDefinition producerMetadataBeanDefinition = producerMetadataBuilder.getBeanDefinition();
 
 			String producerPropertiesBean = parentElem.getAttribute("producer-properties");

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java

@@ -16,12 +16,15 @@
 
 package org.springframework.integration.kafka.support;
 
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
-
+import org.apache.kafka.common.KafkaException;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.GenericConversionService;
 import org.springframework.core.serializer.support.SerializingConverter;
@@ -35,6 +38,7 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @author Marius Bogoevici
  * @author Artem Bilan
+ * @author Martin Dam
  * @since 0.5
  */
 public class ProducerConfiguration<K, V> {
@@ -76,7 +80,25 @@ public class ProducerConfiguration<K, V> {
 
 	public Future<RecordMetadata> send(String topic, Integer partition, K messageKey, V messagePayload) {
 		String targetTopic = StringUtils.hasText(topic) ? topic : this.producerMetadata.getTopic();
-		return this.producer.send(new ProducerRecord<>(targetTopic, partition, messageKey, messagePayload));
+		Future<RecordMetadata> future = this.producer.send(new ProducerRecord<>(targetTopic, partition, messageKey, messagePayload));
+
+		if (!producerMetadata.isSync()) {
+			return future;
+		}
+		else {
+			try {
+				if (producerMetadata.getSendTimeout() <= 0) {
+					future.get();
+				}
+				else {
+					future.get(producerMetadata.getSendTimeout(), TimeUnit.MILLISECONDS);
+				}
+			}
+			catch (InterruptedException | ExecutionException | TimeoutException e) {
+				throw new KafkaException(e);
+			}
+			return future;
+		}
 	}
 
 	public Future<RecordMetadata> convertAndSend(String topic, Integer partition, Object messageKey,

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java

@@ -49,6 +49,10 @@ public class ProducerMetadata<K,V> {
 
 	private int batchBytes = 16384;
 
+	private int sendTimeout = 0;
+
+	private boolean sync = false;
+
 	public ProducerMetadata(final String topic, Class<K> keyClassType, Class<V> valueClassType,
 	                        Serializer<K> keySerializer, Serializer<V> valueSerializer) {
 		Assert.notNull(topic, "Topic cannot be null");
@@ -109,6 +113,22 @@ public class ProducerMetadata<K,V> {
 		return valueClassType;
 	}
 
+	public int getSendTimeout() {
+		return sendTimeout;
+	}
+
+	public void setSendTimeout(int sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
+	public boolean isSync() {
+		return sync;
+	}
+
+	public void setSync(boolean sync) {
+		this.sync = sync;
+	}
+
 	@Override
 	public String toString() {
 		return "ProducerMetadata{" +
@@ -120,6 +140,8 @@ public class ProducerMetadata<K,V> {
 				", partitioner=" + this.partitioner +
 				", compressionType=" + this.compressionType +
 				", batchBytes=" + this.batchBytes +
+				", sync=" + this.sync +
+				", sendTimeout=" + this.sendTimeout +
 				'}';
 	}
 
@@ -128,5 +150,4 @@ public class ProducerMetadata<K,V> {
 		gzip,
 		snappy
 	}
-
 }

spring-integration-kafka/src/main/resources/META-INF/spring.schemas

@@ -1,2 +1,2 @@
-http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.2.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.2.xsd
-http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.2.xsd
+http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.3.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.3.xsd
+http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.3.xsd

spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.3.xsd

@@ -215,6 +215,20 @@
 											</xsd:documentation>
 										</xsd:annotation>
 									</xsd:attribute>
+									<xsd:attribute name="sync" use="optional" type="xsd:boolean">
+										<xsd:annotation>
+											<xsd:documentation>
+												If true, sends messages synchronously. Asynchronously (default) otherwise
+											</xsd:documentation>
+										</xsd:annotation>
+									</xsd:attribute>
+									<xsd:attribute name="send-timeout" use="optional" type="xsd:string">
+										<xsd:annotation>
+											<xsd:documentation>
+												If sync=true, specify timeout in milliseconds. 0 or below indicate forever (default)
+											</xsd:documentation>
+										</xsd:annotation>
+									</xsd:attribute>
 								</xsd:complexType>
 							</xsd:element>
 						</xsd:choice>