GH-1749: Fix DLPR Send Timeout

Resolves #1749

We cannot timeout before the `KafkaProducer`; otherwise we could be reporting
a timeout on a send that actually succeeds.

Ensure that the send timeout is at least a few seconds longer than the producer
timeout.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

@@ -44,7 +44,9 @@
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
@@ -97,6 +99,7 @@ public class DeadLetterPublishingRecoverer implements ConsumerAwareRecordRecover
 
 	private boolean throwIfNoDestinationReturned = false;
 
+	private long timeoutBuffer = Duration.ofSeconds(FIVE).toMillis();
 
 	/**
 	 * Create an instance with the provided template and a default destination resolving
@@ -274,15 +277,28 @@ public void setFailIfSendResultIsError(boolean failIfSendResultIsError) {
 	}
 
 	/**
-	 * Time to wait for message sending. Default is 30 seconds.
+	 * Set the minumum time to wait for message sending. Default is the producer
+	 * configuration {@code delivery.timeout.ms} plus the {@link #setTimeoutBuffer(long)}.
 	 * @param waitForSendResultTimeout the timeout.
 	 * @since 2.7
 	 * @see #setFailIfSendResultIsError(boolean)
+	 * @see #setTimeoutBuffer(long)
 	 */
 	public void setWaitForSendResultTimeout(Duration waitForSendResultTimeout) {
 		this.waitForSendResultTimeout = waitForSendResultTimeout;
 	}
 
+	/**
+	 * Set the number of milliseconds to add to the producer configuration {@code delivery.timeout.ms}
+	 * property to avoid timing out before the Kafka producer. Default 5000.
+	 * @param buffer the buffer.
+	 * @since 2.7
+	 * @see #setWaitForSendResultTimeout(Duration)
+	 */
+	public void setTimeoutBuffer(long buffer) {
+		this.timeoutBuffer = buffer;
+	}
+
 	@SuppressWarnings("unchecked")
 	@Override
 	public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consumer, Exception exception) {
@@ -447,17 +463,20 @@ protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations
 			this.logger.error(e, () -> "Dead-letter publication failed for: " + outRecord);
 		}
 		if (this.failIfSendResultIsError) {
-			verifySendResult(outRecord, sendResult);
+			verifySendResult(kafkaTemplate, outRecord, sendResult);
 		}
 	}
 
-	private void verifySendResult(ProducerRecord<Object, Object> outRecord,
-								@Nullable ListenableFuture<SendResult<Object, Object>> sendResult) {
+	private void verifySendResult(KafkaOperations<Object, Object> kafkaTemplate,
+			ProducerRecord<Object, Object> outRecord,
+			@Nullable ListenableFuture<SendResult<Object, Object>> sendResult) {
+
+		Duration sendTimeout = determineSendTimeout(kafkaTemplate);
 		if (sendResult == null) {
 			throw new KafkaException("Dead-letter publication failed for: " + outRecord);
 		}
 		try {
-			sendResult.get(this.waitForSendResultTimeout.toMillis(), TimeUnit.MILLISECONDS);
+			sendResult.get(sendTimeout.toMillis(), TimeUnit.MILLISECONDS);
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
@@ -468,6 +487,18 @@ private void verifySendResult(ProducerRecord<Object, Object> outRecord,
 		}
 	}
 
+	private Duration determineSendTimeout(KafkaOperations<?, ?> template) {
+		ProducerFactory<? extends Object, ? extends Object> producerFactory = template.getProducerFactory();
+		if (producerFactory != null) { // NOSONAR - will only occur in mock tests
+			Map<String, Object> props = producerFactory.getConfigurationProperties();
+			if (props != null) { // NOSONAR - will only occur in mock tests
+				return KafkaUtils.determineSendTimeout(props, this.timeoutBuffer,
+						this.waitForSendResultTimeout.toMillis());
+			}
+		}
+		return Duration.ofSeconds(THIRTY);
+	}
+
 	private void enhanceHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?> record, Exception exception) {
 		maybeAddOriginalHeaders(kafkaHeaders, record);
 		addExceptionInfoHeaders(kafkaHeaders, exception, false);

spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2021 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.
@@ -19,7 +19,11 @@
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.time.Duration;
 import java.util.Collection;
+import java.util.Map;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
 
 import org.springframework.messaging.Message;
 import org.springframework.util.ClassUtils;
@@ -98,6 +102,39 @@ public static void clearConsumerGroupId() {
 		KafkaUtils.GROUP_IDS.remove();
 	}
 
+	/**
+	 * Return the timeout to use when sending records. If the
+	 * {@link ProducerConfig#DELIVERY_TIMEOUT_MS_CONFIG} is not configured, or is not a
+	 * number or a String that can be parsed as a long, the {@link ProducerConfig} default
+	 * value (plus the buffer) is used.
+	 * @param producerProps the producer properties.
+	 * @param buffer a buffer to add to the configured
+	 * {@link ProducerConfig#DELIVERY_TIMEOUT_MS_CONFIG} to prevent timing out before the
+	 * Kafka producer.
+	 * @param min a minimum value to apply after adding the buffer to the configured
+	 * timeout.
+	 * @return the timeout to use.
+	 * @since 2.7
+	 */
+	public static Duration determineSendTimeout(Map<String, Object> producerProps, long buffer, long min) {
+		Object dt = producerProps.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);
+		if (dt instanceof Number) {
+			return Duration.ofMillis(Math.max(((Number) dt).longValue() + buffer, min));
+		}
+		else if (dt instanceof String) {
+			try {
+				return Duration.ofMillis(Math.max(Long.parseLong((String) dt) + buffer, min));
+			}
+			catch (@SuppressWarnings("unused") NumberFormatException ex) {
+			}
+		}
+		return Duration.ofMillis(Math.max(
+				((Integer) ProducerConfig.configDef().defaultValues()
+						.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG)).longValue() + buffer,
+				min));
+	}
+
+
 	private KafkaUtils() {
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java

@@ -27,6 +27,7 @@
 import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -36,13 +37,15 @@
 import java.io.UncheckedIOException;
 import java.time.Duration;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import org.apache.kafka.clients.consumer.Consumer;
 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.PartitionInfo;
 import org.apache.kafka.common.header.Header;
@@ -56,7 +59,9 @@
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.core.KafkaOperations.OperationsCallback;
+import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
 import org.springframework.util.concurrent.ListenableFuture;
@@ -357,10 +362,15 @@ void replaceOriginalHeaders() {
 		assertThat(anotherHeaders.lastHeader(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE)).isNotEqualTo(originalTimestampType);
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	void failIfSendResultIsError() throws Exception {
 		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
+		ProducerFactory pf = mock(ProducerFactory.class);
+		Map<String, Object> props = new HashMap<>();
+		props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 10L);
+		given(pf.getConfigurationProperties()).willReturn(props);
+		given(template.getProducerFactory()).willReturn(pf);
 		ListenableFuture<?> future = mock(ListenableFuture.class);
 		ArgumentCaptor<Long> timeoutCaptor = ArgumentCaptor.forClass(Long.class);
 		given(template.send(any(ProducerRecord.class))).willReturn(future);
@@ -370,11 +380,61 @@ void failIfSendResultIsError() throws Exception {
 		recoverer.setFailIfSendResultIsError(true);
 		Duration waitForSendResultTimeout = Duration.ofSeconds(1);
 		recoverer.setWaitForSendResultTimeout(waitForSendResultTimeout);
+		recoverer.setTimeoutBuffer(0L);
 		assertThatThrownBy(() -> recoverer.accept(record, new RuntimeException()))
 				.isExactlyInstanceOf(KafkaException.class);
 		assertThat(timeoutCaptor.getValue()).isEqualTo(waitForSendResultTimeout.toMillis());
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void sendTimeoutDefault() throws Exception {
+		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
+		ProducerFactory pf = mock(ProducerFactory.class);
+		Map<String, Object> props = new HashMap<>();
+		given(pf.getConfigurationProperties()).willReturn(props);
+		given(template.getProducerFactory()).willReturn(pf);
+		SettableListenableFuture<SendResult> future = spy(new SettableListenableFuture<>());
+		ArgumentCaptor<Long> timeoutCaptor = ArgumentCaptor.forClass(Long.class);
+		given(template.send(any(ProducerRecord.class))).willReturn(future);
+		willAnswer(inv -> {
+			future.set(new SendResult(null, null));
+			return null;
+		}).given(future).get(timeoutCaptor.capture(), eq(TimeUnit.MILLISECONDS));
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		recoverer.setFailIfSendResultIsError(true);
+		Duration waitForSendResultTimeout = Duration.ofSeconds(1);
+		recoverer.setWaitForSendResultTimeout(waitForSendResultTimeout);
+		recoverer.accept(record, new RuntimeException());
+		assertThat(timeoutCaptor.getValue()).isEqualTo(Duration.ofSeconds(125).toMillis());
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void sendTimeoutConfig() throws Exception {
+		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
+		ProducerFactory pf = mock(ProducerFactory.class);
+		Map<String, Object> props = new HashMap<>();
+		props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 30_000L);
+		given(pf.getConfigurationProperties()).willReturn(props);
+		given(template.getProducerFactory()).willReturn(pf);
+		SettableListenableFuture<SendResult> future = spy(new SettableListenableFuture<>());
+		ArgumentCaptor<Long> timeoutCaptor = ArgumentCaptor.forClass(Long.class);
+		given(template.send(any(ProducerRecord.class))).willReturn(future);
+		willAnswer(inv -> {
+			future.set(new SendResult(null, null));
+			return null;
+		}).given(future).get(timeoutCaptor.capture(), eq(TimeUnit.MILLISECONDS));
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		recoverer.setFailIfSendResultIsError(true);
+		Duration waitForSendResultTimeout = Duration.ofSeconds(1);
+		recoverer.setWaitForSendResultTimeout(waitForSendResultTimeout);
+		recoverer.accept(record, new RuntimeException());
+		assertThat(timeoutCaptor.getValue()).isEqualTo(Duration.ofSeconds(35).toMillis());
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	void notFailIfSendResultIsError() throws Exception {