GH-8638: Kafka: Send All Fails to Failure Channel

Resolves #8638

Previously, immediate failures (e.g. timeout getting metadata) were
only thrown as exceptions, and not sent to the failure channel, if present.

**cherry-pick to all supported branches**
# Conflicts:
#	spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2021 the original author or authors.
+ * Copyright 2013-2023 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.
@@ -423,6 +423,7 @@ public String getComponentType() {
 		return this.isGateway ? "kafka:outbound-gateway" : "kafka:outbound-channel-adapter";
 	}
 
+	@Nullable
 	protected MessageChannel getSendFailureChannel() {
 		if (this.sendFailureChannel != null) {
 			return this.sendFailureChannel;
@@ -500,19 +501,27 @@ protected Object handleRequestMessage(final Message<?> message) {
 		}
 		ListenableFuture<SendResult<K, V>> sendFuture;
 		RequestReplyFuture<K, V, Object> gatewayFuture = null;
-		if (this.isGateway && (!preBuilt || producerRecord.headers().lastHeader(KafkaHeaders.REPLY_TOPIC) == null)) {
-			producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message)));
-			gatewayFuture = ((ReplyingKafkaTemplate<K, V, Object>) this.kafkaTemplate).sendAndReceive(producerRecord);
-			sendFuture = gatewayFuture.getSendFuture();
-		}
-		else {
-			if (this.transactional && !this.kafkaTemplate.inTransaction() && !this.allowNonTransactional) {
-				sendFuture = this.kafkaTemplate.executeInTransaction(template -> template.send(producerRecord));
+		try {
+			if (this.isGateway
+					&& (!preBuilt || producerRecord.headers().lastHeader(KafkaHeaders.REPLY_TOPIC) == null)) {
+				producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message)));
+				gatewayFuture = ((ReplyingKafkaTemplate<K, V, Object>) this.kafkaTemplate)
+						.sendAndReceive(producerRecord);
+				sendFuture = gatewayFuture.getSendFuture();
 			}
 			else {
-				sendFuture = this.kafkaTemplate.send(producerRecord);
+				if (this.transactional && !this.kafkaTemplate.inTransaction() && !this.allowNonTransactional) {
+					sendFuture = this.kafkaTemplate.executeInTransaction(template -> template.send(producerRecord));
+				}
+				else {
+					sendFuture = this.kafkaTemplate.send(producerRecord);
+				}
 			}
 		}
+		catch (RuntimeException rtex) {
+			sendFailure(message, producerRecord, getSendFailureChannel(), rtex);
+			throw rtex;
+		}
 		sendFutureIfRequested(sendFuture, futureToken);
 		if (flush) {
 			this.kafkaTemplate.flush();
@@ -680,11 +689,7 @@ public void onSuccess(SendResult<K, V> result) {
 
 				@Override
 				public void onFailure(Throwable ex) {
-					if (failureChannel != null) {
-						KafkaProducerMessageHandler.this.messagingTemplate.send(failureChannel,
-								KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage(
-										new KafkaSendFailureException(message, producerRecord, ex), null));
-					}
+					sendFailure(message, producerRecord, failureChannel, ex);
 				}
 
 			});
@@ -713,6 +718,16 @@ public void onFailure(Throwable ex) {
 		}
 	}
 
+	private void sendFailure(final Message<?> message, final ProducerRecord<K, V> producerRecord,
+			@Nullable MessageChannel failureChannel, Throwable exception) {
+
+		if (failureChannel != null) {
+			KafkaProducerMessageHandler.this.messagingTemplate.send(failureChannel,
+					KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage(
+							new KafkaSendFailureException(message, producerRecord, exception), null));
+		}
+	}
+
 	private Future<?> processReplyFuture(@Nullable RequestReplyFuture<?, ?, Object> future) {
 		if (future == null) {
 			return null;

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

@@ -59,6 +59,7 @@
 import org.springframework.integration.kafka.support.KafkaIntegrationHeaders;
 import org.springframework.integration.kafka.support.KafkaSendFailureException;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -94,6 +95,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.InstanceOfAssertFactories.throwable;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
@@ -343,6 +345,35 @@ protected ListenableFuture<SendResult<Integer, String>> doSend(
 		producerFactory.destroy();
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void immediateFailure() {
+		Producer producer = mock(Producer.class);
+		CompletableFuture cf = new CompletableFuture();
+		RuntimeException rte = new RuntimeException("test.immediate");
+		cf.completeExceptionally(rte);
+		given(producer.send(any(), any())).willReturn(cf);
+		ProducerFactory pf = mock(ProducerFactory.class);
+		given(pf.createProducer()).willReturn(producer);
+		KafkaTemplate template = new KafkaTemplate(pf);
+		template.setDefaultTopic("foo");
+		KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template);
+		QueueChannel fails = new QueueChannel();
+		handler.setSendFailureChannel(fails);
+		assertThatExceptionOfType(MessageHandlingException.class).isThrownBy(
+						() -> handler.handleMessage(new GenericMessage<>("")))
+				.withCauseExactlyInstanceOf(KafkaException.class)
+				.withStackTraceContaining("test.immediate");
+		Message<?> fail = fails.receive(0);
+		assertThat(fail).isNotNull();
+		assertThat(fail.getPayload())
+			.asInstanceOf(throwable(KafkaSendFailureException.class))
+			.cause()
+			.isInstanceOf(KafkaException.class)
+			.cause()
+			.isEqualTo(rte);
+	}
+
 	@Test
 	void testOutboundWithCustomHeaderMapper() {
 		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(

src/checkstyle/checkstyle.xml

@@ -80,6 +80,7 @@
 					  value="org.assertj.core.api.Assertions.*,
 					  org.xmlunit.assertj3.XmlAssert.*,
 				org.assertj.core.api.Assumptions.*,
+				org.assertj.core.api.InstanceOfAssertFactories.*,
 				org.awaitility.Awaitility.*,
 				org.mockito.Mockito.*,
 				org.mockito.BDDMockito.*,