GH-2419: DLPR: Protect Against Non-Compliant PF

Resolves #2419

We try to obtain the configured request timeout using the producer factory's
configuration properties. Some wrappers (e.g. open tracing) do not implement
this method, so fallback to the default instead of throwing an USOE.

**cherry-pick to 2.9.x, 2.8.x**

* Fix `CompletableFuture` conflicts in the tests

Author: garyrussell

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

@@ -703,7 +703,13 @@ private String pubFailMessage(ProducerRecord<Object, Object> outRecord, Consumer
 	protected 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();
+			Map<String, Object> props;
+			try {
+				props = producerFactory.getConfigurationProperties();
+			}
+			catch (UnsupportedOperationException ex) {
+				props = Collections.emptyMap();
+			}
 			if (props != null) { // NOSONAR - will only occur in mock tests
 				return KafkaUtils.determineSendTimeout(props, this.timeoutBuffer,
 						this.waitForSendResultTimeout.toMillis());

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

@@ -24,6 +24,7 @@
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.then;
 import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.BDDMockito.willCallRealMethod;
 import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
@@ -315,7 +316,7 @@ private byte[] header(boolean isKey, DeserializationException deserEx) {
 		return baos.toByteArray();
 	}
 
-	@SuppressWarnings({"unchecked", "rawtypes"})
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	void allOriginalHeaders() {
 		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
@@ -335,7 +336,7 @@ void allOriginalHeaders() {
 		assertThat(headers.lastHeader(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE)).isNotNull();
 	}
 
-	@SuppressWarnings({"unchecked", "rawtypes"})
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	void dontAppendOriginalHeaders() {
 		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
@@ -386,7 +387,7 @@ void dontAppendOriginalHeaders() {
 		assertThat(exceptionHeaders.hasNext()).isFalse();
 	}
 
-	@SuppressWarnings({"unchecked", "rawtypes"})
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	void appendOriginalHeaders() {
 		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
@@ -825,13 +826,13 @@ void addHeaderFunctionsProcessedInOrder() {
 		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
 		recoverer.setHeadersFunction((rec, ex) -> {
-			return new RecordHeaders(new RecordHeader[] { new RecordHeader("foo", "one".getBytes()) });
+			return new RecordHeaders(new RecordHeader[]{ new RecordHeader("foo", "one".getBytes()) });
 		});
 		recoverer.addHeadersFunction((rec, ex) -> {
-			return new RecordHeaders(new RecordHeader[] { new RecordHeader("bar", "two".getBytes()) });
+			return new RecordHeaders(new RecordHeader[]{ new RecordHeader("bar", "two".getBytes()) });
 		});
 		recoverer.addHeadersFunction((rec, ex) -> {
-			return new RecordHeaders(new RecordHeader[] { new RecordHeader("foo", "three".getBytes()) });
+			return new RecordHeaders(new RecordHeader[]{ new RecordHeader("foo", "three".getBytes()) });
 		});
 		recoverer.accept(record, new ListenerExecutionFailedException("test", "group", new RuntimeException()));
 		ArgumentCaptor<ProducerRecord> producerRecordCaptor = ArgumentCaptor.forClass(ProducerRecord.class);
@@ -862,12 +863,12 @@ void immutableHeaders() {
 		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
 		recoverer.setHeadersFunction((rec, ex) -> {
-			RecordHeaders headers = new RecordHeaders(new RecordHeader[] { new RecordHeader("foo", "one".getBytes()) });
+			RecordHeaders headers = new RecordHeaders(new RecordHeader[]{ new RecordHeader("foo", "one".getBytes()) });
 			headers.setReadOnly();
 			return headers;
 		});
 		recoverer.addHeadersFunction((rec, ex) -> {
-			return new RecordHeaders(new RecordHeader[] { new RecordHeader("bar", "two".getBytes()) });
+			return new RecordHeaders(new RecordHeader[]{ new RecordHeader("bar", "two".getBytes()) });
 		});
 		recoverer.accept(record, new ListenerExecutionFailedException("test", "group", new RuntimeException()));
 		ArgumentCaptor<ProducerRecord> producerRecordCaptor = ArgumentCaptor.forClass(ProducerRecord.class);
@@ -877,4 +878,25 @@ void immutableHeaders() {
 		assertThat(KafkaTestUtils.getPropertyValue(headers, "headers", List.class)).hasSize(12);
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	void nonCompliantProducerFactory() throws Exception {
+		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
+		ProducerFactory pf = mock(ProducerFactory.class);
+
+		willCallRealMethod().given(pf).getConfigurationProperties();
+
+		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);
+		given(future.get(timeoutCaptor.capture(), eq(TimeUnit.MILLISECONDS))).willThrow(new TimeoutException());
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		recoverer.setFailIfSendResultIsError(true);
+		assertThatThrownBy(() -> recoverer.accept(record, new RuntimeException()))
+				.isExactlyInstanceOf(KafkaException.class);
+		assertThat(timeoutCaptor.getValue()).isEqualTo(Duration.ofSeconds(125).toMillis());
+	}
+
 }

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

@@ -41,6 +41,7 @@
 
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaOperations;