GH-125: Introduce sync mode for KafkaProducerMH

Fixes: GH-125 (https://github.com/spring-projects/spring-integration-kafka/issues/125)

* Add `sync` mode for the `KafkaProducerMessageHandler` (`false` by default) to wait for result from the send `Future`
* Add `sendTimeout` do not block `Future.get()` forever
* Provide test-case based on the `MockProducer` to complete the record with an exception and verify that the `sync` mode works well
* Some dependencies upgrade

Wrap `Future.get()` `TimeoutException` into `MessageTimeoutException`

Fix whitespace after comma

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 the original author or authors.
+ * Copyright 2013-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,8 +16,12 @@
 
 package org.springframework.integration.kafka.outbound;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
+import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -25,6 +29,7 @@ import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
+import org.springframework.util.concurrent.ListenableFuture;
 
 /**
  * Kafka Message Handler.
@@ -40,6 +45,8 @@ import org.springframework.util.StringUtils;
  */
 public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
 
+	private static final long DEFAULT_SEND_TIMEOUT = 10000;
+
 	private final KafkaTemplate<K, V> kafkaTemplate;
 
 	private EvaluationContext evaluationContext;
@@ -50,6 +57,10 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
 
 	private volatile Expression partitionIdExpression;
 
+	private boolean sync;
+
+	private long sendTimeout = DEFAULT_SEND_TIMEOUT;
+
 	public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
 		Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null");
 		this.kafkaTemplate = kafkaTemplate;
@@ -71,6 +82,26 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
 		return this.kafkaTemplate;
 	}
 
+	/**
+	 * The {@code boolean} indicated if {@link KafkaProducerMessageHandler}
+	 * should wait for send operation results or not. Defaults to {@code false}.
+	 * In {@code sync} mode a downstream send operation exception will be re-thrown.
+	 * @param sync the send mode; async by default.
+	 * @since 2.0.1
+	 */
+	public void setSync(boolean sync) {
+		this.sync = sync;
+	}
+
+	/**
+	 * Specify a timeout in milliseconds how long {@link KafkaProducerMessageHandler}
+	 * should wait wait for send operation results. Defaults to 10 seconds.
+	 * @param sendTimeout the timeout to wait for result fo send operation.
+	 */
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		super.onInit();
@@ -94,20 +125,35 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
 				? this.messageKeyExpression.getValue(this.evaluationContext, message)
 				: message.getHeaders().get(KafkaHeaders.MESSAGE_KEY);
 
+		ListenableFuture<?> future;
+
 		if (partitionId == null) {
 			if (messageKey == null) {
-				this.kafkaTemplate.send(topic, (V) message.getPayload());
+				future = this.kafkaTemplate.send(topic, (V) message.getPayload());
 			}
 			else {
-				this.kafkaTemplate.send(topic, (K) messageKey, (V) message.getPayload());
+				future = this.kafkaTemplate.send(topic, (K) messageKey, (V) message.getPayload());
 			}
 		}
 		else {
 			if (messageKey == null) {
-				this.kafkaTemplate.send(topic, partitionId, (V) message.getPayload());
+				future = this.kafkaTemplate.send(topic, partitionId, (V) message.getPayload());
 			}
 			else {
-				this.kafkaTemplate.send(topic, partitionId, (K) messageKey, (V) message.getPayload());
+				future = this.kafkaTemplate.send(topic, partitionId, (K) messageKey, (V) message.getPayload());
+			}
+		}
+		if (this.sync) {
+			if (this.sendTimeout < 0) {
+				future.get();
+			}
+			else {
+				try {
+					future.get(this.sendTimeout, TimeUnit.MILLISECONDS);
+				}
+				catch (TimeoutException te) {
+					throw new MessageTimeoutException(message, "Timeout waiting for response from KafkaProducer", te);
+				}
 			}
 		}
 	}

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java

@@ -17,14 +17,29 @@
 package org.springframework.integration.kafka.config.xml;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.Mockito.mock;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -54,4 +69,40 @@ public class KafkaOutboundAdapterParserTests {
 		assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.expression")).isEqualTo("2");
 	}
 
+
+	@Test
+	public void testSyncMode() {
+		MockProducer<Integer, String> mockProducer =
+				new MockProducer<>(false, new IntegerSerializer(), new StringSerializer());
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(() -> mockProducer);
+		KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.afterPropertiesSet();
+
+		handler.setSync(true);
+		handler.setTopicExpression(new LiteralExpression("foo"));
+
+		Executors.newSingleThreadExecutor()
+				.submit(() -> {
+					RuntimeException exception = new RuntimeException("Async Producer Mock exception");
+					while (!mockProducer.errorNext(exception)) {
+						Thread.sleep(100);
+					}
+					return null;
+				});
+
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo")))
+				.withMessageContaining("Async Producer Mock exception")
+				.withCauseExactlyInstanceOf(ExecutionException.class)
+				.withRootCauseExactlyInstanceOf(RuntimeException.class);
+
+		handler.setSendTimeout(1);
+
+		assertThatExceptionOfType(MessageTimeoutException.class)
+				.isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo")))
+				.withMessageContaining("Timeout waiting for response from KafkaProducer")
+				.withCauseExactlyInstanceOf(TimeoutException.class);
+	}
+
 }