From c4ee5512f6f7170ada43af4f1167d74adb2e0dd2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 8 Jun 2023 13:44:29 -0400 Subject: [PATCH] GH-8638: Kafka: Send All Fails to Failure Channel Resolves https://github.com/spring-projects/spring-integration/issues/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** --- .../outbound/KafkaProducerMessageHandler.java | 45 +++++++++++++------ .../KafkaProducerMessageHandlerTests.java | 33 +++++++++++++- src/checkstyle/checkstyle.xml | 1 + 3 files changed, 64 insertions(+), 15 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index f6eaacf233..961596e0d0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 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. @@ -438,6 +438,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes return this.isGateway ? "kafka:outbound-gateway" : "kafka:outbound-channel-adapter"; } + @Nullable protected MessageChannel getSendFailureChannel() { if (this.sendFailureChannel != null) { return this.sendFailureChannel; @@ -515,19 +516,27 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } CompletableFuture> sendFuture; RequestReplyFuture 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) 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) 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(); @@ -699,10 +708,8 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes .build()); } } - else if (failureChannel != null) { - KafkaProducerMessageHandler.this.messagingTemplate.send(failureChannel, - KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage( - new KafkaSendFailureException(message, producerRecord, exception), null)); + else { + sendFailure(message, producerRecord, failureChannel, exception); } }); } @@ -730,6 +737,16 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } } + private void sendFailure(final Message message, final ProducerRecord 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 future) { if (future == null) { return null; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 54759d6d85..28317ab32d 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2022 the original author or authors. + * Copyright 2016-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. @@ -60,6 +60,7 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle 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; @@ -93,6 +94,7 @@ import org.springframework.transaction.support.TransactionTemplate; 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; @@ -344,6 +346,35 @@ class KafkaProducerMessageHandlerTests { 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 producerFactory = new DefaultKafkaProducerFactory<>( diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index f557a20d5e..0a23156ce3 100644 --- a/src/checkstyle/checkstyle.xml +++ b/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.*,