Fix Kafka Send Timeout

Resolves spring-attic/spring-cloud-stream-binder-kafka#928

Kafka has a longer default send timeout; this means a send could be successful long
after Spring has timed out the send.

**I will backport to the 3.3.x extension after merge**

* Use setSendTimeout; make it final.

Author: garyrussell

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

@@ -30,6 +30,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Headers;
@@ -101,7 +102,10 @@
 public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMessageHandler
 		implements Lifecycle {
 
-	private static final long DEFAULT_SEND_TIMEOUT = 10000;
+	/**
+	 * Buffer added to ensure our timeout is longer than Kafka's.
+	 */
+	private static final int TIMEOUT_BUFFER = 5000;
 
 	private final Map<String, Set<Integer>> replyTopicsAndPartitions = new HashMap<>();
 
@@ -115,6 +119,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 	private final AtomicBoolean running = new AtomicBoolean();
 
+	private final long deliveryTimeoutMsProperty;
+
 	private EvaluationContext evaluationContext;
 
 	private Expression topicExpression;
@@ -130,7 +136,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 	private boolean sync;
 
-	private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT);
+	private Expression sendTimeoutExpression;
 
 	private KafkaHeaderMapper headerMapper;
 
@@ -175,6 +181,25 @@ public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
 			logger.warn("The KafkaTemplate is transactional; this gateway will only work if the consumer is "
 					+ "configured to read uncommitted records");
 		}
+		determineSendTimeout();
+		this.deliveryTimeoutMsProperty = this.sendTimeoutExpression.getValue(Long.class) - TIMEOUT_BUFFER;
+	}
+
+	private void determineSendTimeout() {
+		Map<String, Object> props = this.kafkaTemplate.getProducerFactory().getConfigurationProperties();
+		Object dt = props.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);
+		if (dt == null) {
+			dt = ProducerConfig.configDef().defaultValues().get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);
+		}
+		if (dt instanceof Long) {
+			setSendTimeout(((Long) dt) + TIMEOUT_BUFFER);
+		}
+		else if (dt instanceof Integer) {
+			setSendTimeout(Long.valueOf((Integer) dt) + TIMEOUT_BUFFER);
+		}
+		else if (dt instanceof String) {
+			setSendTimeout(Long.parseLong((String) dt) + TIMEOUT_BUFFER);
+		}
 	}
 
 	public void setTopicExpression(Expression topicExpression) {
@@ -243,24 +268,25 @@ public void setSync(boolean sync) {
 
 	/**
 	 * Specify a timeout in milliseconds for how long this
-	 * {@link KafkaProducerMessageHandler} should wait wait for send operation
-	 * results. Defaults to 10 seconds. The timeout is applied only in {@link #sync} mode.
-	 * Also applies when sending to the success or failure channels.
-	 * @param sendTimeout the timeout to wait for result fo send operation.
+	 * {@link KafkaProducerMessageHandler} should wait wait for send operation results.
+	 * Defaults to the kafka {@code delivery.timeout.ms} property + 5 seconds. The timeout
+	 * is applied Also applies when sending to the success or failure channels.
+	 * @param sendTimeout the timeout to wait for result for a send operation.
 	 * @since 2.0.1
 	 */
 	@Override
-	public void setSendTimeout(long sendTimeout) {
+	public final void setSendTimeout(long sendTimeout) {
 		super.setSendTimeout(sendTimeout);
 		setSendTimeoutExpression(new ValueExpression<>(sendTimeout));
 	}
 
 	/**
 	 * Specify a SpEL expression to evaluate a timeout in milliseconds for how long this
-	 * {@link KafkaProducerMessageHandler} should wait wait for send operation
-	 * results. Defaults to 10 seconds. The timeout is applied only in {@link #sync} mode.
+	 * {@link KafkaProducerMessageHandler} should wait wait for send operation results.
+	 * Defaults to the kafka {@code delivery.timeout.ms} property + 5 seconds. The timeout
+	 * is applied only in {@link #sync} mode.
 	 * @param sendTimeoutExpression the {@link Expression} for timeout to wait for result
-	 * fo send operation.
+	 * for a send operation.
 	 * @since 2.1.1
 	 */
 	public void setSendTimeoutExpression(Expression sendTimeoutExpression) {
@@ -591,8 +617,15 @@ public void onFailure(Throwable ex) {
 			});
 		}
 
-		if (this.sync) {
+		if (this.sync || this.isGateway) {
 			Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class);
+			if (sendTimeout != null && sendTimeout <= this.deliveryTimeoutMsProperty) {
+				this.logger.debug("'sendTimeout' increased to "
+						+ (this.deliveryTimeoutMsProperty + TIMEOUT_BUFFER)
+						+ "ms; it must be greater than the 'delivery.timeout.ms' Kafka producer "
+						+ "property to avoid false failures");
+				sendTimeout = this.deliveryTimeoutMsProperty + TIMEOUT_BUFFER;
+			}
 			if (sendTimeout == null || sendTimeout < 0) {
 				future.get();
 			}

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

@@ -38,9 +38,6 @@
 
 	<bean id="ems" class="org.springframework.integration.kafka.config.xml.KafkaOutboundGatewayParserTests$EMS"/>
 
-	<bean id="template" class="org.mockito.Mockito" factory-method="mock">
-		<constructor-arg value="org.springframework.kafka.requestreply.ReplyingKafkaTemplate"/>
-	</bean>
+	<bean id="customHeaderMapper" class="org.springframework.kafka.support.DefaultKafkaHeaderMapper"/>
 
-	<bean id="customHeaderMapper" class="org.springframework.kafka.support.DefaultKafkaHeaderMapper" />
 </beans>

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

@@ -17,15 +17,24 @@
 package org.springframework.integration.kafka.config.xml;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+import java.util.HashMap;
+import java.util.Map;
 
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
 import org.springframework.integration.support.DefaultErrorMessageStrategy;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
@@ -74,8 +83,20 @@ public void testProps() {
 				.isSameAs(this.context.getBean("customHeaderMapper"));
 	}
 
+	@Component
 	public static class EMS extends DefaultErrorMessageStrategy {
 
+		@SuppressWarnings("rawtypes")
+		@Bean
+		public KafkaTemplate template() {
+			ProducerFactory pf = mock(ProducerFactory.class);
+			Map<String, Object> props = new HashMap<>();
+			given(pf.getConfigurationProperties()).willReturn(props);
+			KafkaTemplate template = mock(KafkaTemplate.class);
+			given(template.getProducerFactory()).willReturn(pf);
+			return template;
+		}
+
 	}
 
 }

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java

@@ -54,6 +54,7 @@
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -147,11 +148,14 @@ static void tearDown() {
 
 	@Test
 	void testOutbound() {
-		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
-				KafkaTestUtils.producerProps(embeddedKafka));
+		Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
+		producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 40_000);
+		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(producerProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
 		KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
 		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setSendTimeout(50_000);
+		handler.setSync(true);
 		handler.afterPropertiesSet();
 
 		Message<?> message = MessageBuilder.withPayload("foo")

src/reference/asciidoc/kafka.adoc

@@ -73,6 +73,10 @@ Flushing after sending several messages might be useful if you are using the `li
 By default, the expression looks for a `Boolean` value in the `KafkaIntegrationHeaders.FLUSH` header (`kafka_flush`).
 The flush will occur if the value is `true` and not if it's `false` or the header is absent.
 
+Starting with version 5.4, the `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed from 10 seconds to the `delivery.timeout.ms` Kafka producer property `+ 5000` so that the actual Kafka error after a timeout is propagated to the application, instead of a timeout generated by this framework.
+This has been changed for consistency because you may get unexpected behavior (Spring may timeout the send, while it is actually, eventually, successful).
+IMPORTANT: That timeout is 120 seconds by default so you may wish to reduce it to get more timely failures.
+
 ==== Java Configuration
 
 The following example shows how to configure the Kafka outbound channel adapter with Java:
@@ -425,6 +429,10 @@ If your code invokes the gateway behind a synchronous https://docs.spring.io/spr
 IMPORTANT: The gateway does not accept requests until the reply container has been assigned its topics and partitions.
 It is suggested that you add a `ConsumerRebalanceListener` to the template's reply container properties and wait for the `onPartitionsAssigned` call before sending messages to the gateway.
 
+Starting with version 5.4, the `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed from 10 seconds to the `delivery.timeout.ms` Kafka producer property `+ 5000` so that the actual Kafka error after a timeout is propagated to the application, instead of a timeout generated by this framework.
+This has been changed for consistency because you may get unexpected behavior (Spring may timeout the send, while it is actually, eventually, successful).
+IMPORTANT: That timeout is 120 seconds by default so you may wish to reduce it to get more timely failures.
+
 ==== Java Configuration
 
 The following example shows how to configure a gateway with Java:

src/reference/asciidoc/whats-new.adoc

@@ -20,6 +20,9 @@ If you are interested in more details, see the Issue Tracker tickets that were r
 The standalone https://projects.spring.io/spring-integration-kafka/[Spring Integration Kafka] project has been merged as a `spring-integration-kafka` module to this project.
 See <<./kafka.adoc#kafka,Spring for Apache Kafka Support>> for more information.
 
+The `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed.
+See <<./kafka.adoc#kafka-outbound,Kafka Outbound Channel Adapter>> for more information.
+
 ==== R2DBC Channel Adapters
 
 The Channel Adapters for R2DBC database interaction have been introduced.