From 19c3c4607f20b6622195c924e4d8a3e29120d3ff Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 4 Oct 2018 12:09:15 -0400 Subject: [PATCH] Fix TODOs in outbound channel adapter - always use `successChannel` for metadata - unwrap `ExecutionException` --- .../xml/KafkaOutboundChannelAdapterParser.java | 5 ++--- .../outbound/KafkaProducerMessageHandler.java | 17 ++--------------- .../xml/KafkaOutboundAdapterParserTests.java | 8 ++++---- .../KafkaProducerMessageHandlerTests.java | 2 +- 4 files changed, 9 insertions(+), 23 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java index b59a0c8e68..29f1d2f3a9 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -85,8 +85,7 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-failure-channel"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-success-channel", - "outputChannel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-success-channel"); return builder.getBeanDefinition(); } 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 e4a3c14c78..28ae87729a 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 @@ -367,9 +367,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes (K) messageKey, payload, headers); ListenableFuture> sendFuture; RequestReplyFuture gatewayFuture = null; - MessageChannel metadataChannel = null; if (this.isGateway) { - metadataChannel = getSendSuccessChannel(); producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message))); gatewayFuture = ((ReplyingKafkaTemplate) this.kafkaTemplate).sendAndReceive(producerRecord); sendFuture = gatewayFuture.getSendFuture(); @@ -383,27 +381,16 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes else { sendFuture = this.kafkaTemplate.send(producerRecord); } - // TODO: In 3.1, always use the success channel. - if (!this.noOutputChannel) { - metadataChannel = getOutputChannel(); - if (metadataChannel == null) { - this.noOutputChannel = true; - } - } - if (metadataChannel == null) { - metadataChannel = getSendSuccessChannel(); - } } try { - processSendResult(message, producerRecord, sendFuture, metadataChannel); + processSendResult(message, producerRecord, sendFuture, getSendSuccessChannel()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new MessageHandlingException(message, e); } catch (ExecutionException e) { - // TODO: in 3.1 change this to e.getCause() - throw new MessageHandlingException(message, e); + throw new MessageHandlingException(message, e.getCause()); } return processReplyFuture(gatewayFuture); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index 4fc614d234..d281ddeafe 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -20,7 +20,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.TimeoutException; @@ -37,6 +36,7 @@ import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.test.util.TestUtils; +import org.springframework.kafka.core.KafkaProducerException; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.support.GenericMessage; @@ -78,7 +78,7 @@ public class KafkaOutboundAdapterParserTests { .isSameAs(this.appContext.getBean("ems")); assertThat(TestUtils.getPropertyValue(messageHandler, "sendFailureChannel")) .isSameAs(this.appContext.getBean("failures")); - assertThat(TestUtils.getPropertyValue(messageHandler, "outputChannel")) + assertThat(TestUtils.getPropertyValue(messageHandler, "sendSuccessChannel")) .isSameAs(this.appContext.getBean("successes")); messageHandler @@ -125,7 +125,7 @@ public class KafkaOutboundAdapterParserTests { } catch (Exception e) { assertThat(e).isInstanceOf(MessageHandlingException.class); - assertThat(e.getCause()).isExactlyInstanceOf(ExecutionException.class); + assertThat(e.getCause()).isExactlyInstanceOf(KafkaProducerException.class); assertThat(e.getCause().getCause()).isInstanceOf(RuntimeException.class); assertThat(e.getMessage()).contains("Async Producer Mock exception"); } 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 6e8cb3b3a0..1636c12dd9 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 @@ -256,7 +256,7 @@ public class KafkaProducerMessageHandlerTests { KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); handler.setBeanFactory(mock(BeanFactory.class)); PollableChannel successes = new QueueChannel(); - handler.setOutputChannel(successes); + handler.setSendSuccessChannel(successes); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload("foo")