From 573c29793a76c2d2d7c5d284e82691488aaf1b9d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 25 Jul 2022 16:44:02 -0400 Subject: [PATCH] Fix SIK module for latest SK compatibility Spring for Apache Kafka does not produce `ListenableFuture` anymore. * Rework `KafkaProducerMessageHandler` to deal with the `CompletableFuture` from now on * Add support for `CompletableFuture` replies handling into `AbstractMessageProducingHandler` * Remove redundant cast in the `DispatcherHasNoSubscribersTests` --- .../DispatcherHasNoSubscribersTests.java | 4 +- .../AbstractMessageProducingHandler.java | 70 +++++++------- .../outbound/KafkaProducerMessageHandler.java | 96 ++++++++----------- .../KafkaProducerMessageHandlerTests.java | 10 +- 4 files changed, 88 insertions(+), 92 deletions(-) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java index abbc8ab226..7ca572a1ff 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java @@ -77,7 +77,7 @@ public class DispatcherHasNoSubscribersTests { amqpChannel.setBeanFactory(mock(BeanFactory.class)); amqpChannel.afterPropertiesSet(); - MessageListener listener = (MessageListener) container.getMessageListener(); + MessageListener listener = container.getMessageListener(); assertThatExceptionOfType(MessageDeliveryException.class) .isThrownBy(() -> listener.onMessage(new Message("Hello world!".getBytes()))) @@ -101,7 +101,7 @@ public class DispatcherHasNoSubscribersTests { amqpChannel.afterPropertiesSet(); List logList = insertMockLoggerInListener(amqpChannel); - MessageListener listener = (MessageListener) container.getMessageListener(); + MessageListener listener = container.getMessageListener(); listener.onMessage(new Message("Hello world!".getBytes())); verifyLogReceived(logList); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index b6b8481696..af2416642d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2021 the original author or authors. + * Copyright 2014-2022 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. @@ -23,7 +23,9 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; import org.reactivestreams.Publisher; @@ -50,8 +52,6 @@ import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; -import org.springframework.util.concurrent.SettableListenableFuture; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -307,7 +307,10 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan replyChannel = getOutputChannel(); } - if (this.async && (reply instanceof ListenableFuture || reply instanceof Publisher)) { + if (this.async && (reply instanceof ListenableFuture + || reply instanceof CompletableFuture + || reply instanceof Publisher)) { + if (reply instanceof Publisher && replyChannel instanceof ReactiveStreamsSubscribableChannel) { @@ -349,12 +352,14 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } private void asyncNonReactiveReply(Message requestMessage, Object reply, @Nullable Object replyChannel) { - ListenableFuture future; - if (reply instanceof ListenableFuture) { - future = (ListenableFuture) reply; + CompletableFuture future; + if (reply instanceof CompletableFuture) { + future = (CompletableFuture) reply; + } + else if (reply instanceof ListenableFuture) { + future = ((ListenableFuture) reply).completable(); } else { - SettableListenableFuture settableListenableFuture = new SettableListenableFuture<>(); Mono reactiveReply; ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply); if (adapter != null && adapter.isMultiValue()) { @@ -363,12 +368,10 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan else { reactiveReply = Mono.from((Publisher) reply); } - reactiveReply - .publishOn(Schedulers.boundedElastic()) - .subscribe(settableListenableFuture::set, settableListenableFuture::setException); - future = settableListenableFuture; + + future = reactiveReply.publishOn(Schedulers.boundedElastic()).toFuture(); } - future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel)); + future.whenComplete(new ReplyFutureCallback(requestMessage, replyChannel)); } private Object getOutputChannelFromRoutingSlip(Object reply, Message requestMessage, List routingSlip, @@ -517,7 +520,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return errorChannel; } - private final class ReplyFutureCallback implements ListenableFutureCallback { + private final class ReplyFutureCallback implements BiConsumer { private final Message requestMessage; @@ -529,29 +532,32 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan this.replyChannel = replyChannel; } - @Override - public void onSuccess(Object result) { - Message replyMessage = null; - try { - replyMessage = createOutputMessage(result, this.requestMessage.getHeaders()); - sendOutput(replyMessage, this.replyChannel, false); - } - catch (Exception ex) { - Exception exceptionToLogAndSend = ex; - if (!(ex instanceof MessagingException)) { // NOSONAR - exceptionToLogAndSend = new MessageHandlingException(this.requestMessage, ex); - if (replyMessage != null) { - exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend); - } + public void accept(Object result, Throwable exception) { + if (exception == null) { + Message replyMessage = null; + try { + replyMessage = createOutputMessage(result, this.requestMessage.getHeaders()); + sendOutput(replyMessage, this.replyChannel, false); } - logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + result.toString()); - onFailure(exceptionToLogAndSend); + catch (Exception ex) { + Exception exceptionToLogAndSend = ex; + if (!(ex instanceof MessagingException)) { // NOSONAR + exceptionToLogAndSend = new MessageHandlingException(this.requestMessage, ex); + if (replyMessage != null) { + exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend); + } + } + logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + result.toString()); + onFailure(exceptionToLogAndSend); + } + } + else { + onFailure(exception); } } - @Override - public void onFailure(Throwable ex) { + private void onFailure(Throwable ex) { sendErrorMessage(this.requestMessage, ex); } 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 801b55cad1..cc3c578ad7 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 @@ -23,13 +23,13 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; import java.util.TreeSet; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; -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.TopicPartition; @@ -69,9 +69,6 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; -import org.springframework.util.concurrent.SettableListenableFuture; /** * A Message Handler for Apache Kafka; when supplied with a {@link ReplyingKafkaTemplate} it is used as @@ -174,7 +171,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes if (this.isGateway) { setAsync(true); updateNotPropagatedHeaders( - new String[]{KafkaHeaders.TOPIC, KafkaHeaders.PARTITION, KafkaHeaders.KEY}, false); + new String[]{ KafkaHeaders.TOPIC, KafkaHeaders.PARTITION, KafkaHeaders.KEY }, false); } if (JacksonPresent.isJackson2Present()) { this.headerMapper = new DefaultKafkaHeaderMapper(); @@ -498,7 +495,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes if (futureToken != null) { producerRecord.headers().remove(KafkaIntegrationHeaders.FUTURE_TOKEN); } - ListenableFuture> sendFuture; + 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))); @@ -530,7 +527,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes return processReplyFuture(gatewayFuture); } - private void sendFutureIfRequested(ListenableFuture> sendFuture, Object futureToken) { + private void sendFutureIfRequested(CompletableFuture> sendFuture, Object futureToken) { if (futureToken != null) { MessageChannel futures = getFuturesChannel(); @@ -662,31 +659,26 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } public void processSendResult(final Message message, final ProducerRecord producerRecord, - ListenableFuture> future, MessageChannel metadataChannel) + CompletableFuture> future, MessageChannel metadataChannel) throws InterruptedException, ExecutionException { final MessageChannel failureChannel = getSendFailureChannel(); if (failureChannel != null || metadataChannel != null) { - future.addCallback(new ListenableFutureCallback>() { // NOSONAR - - @Override - public void onSuccess(SendResult result) { + future.whenComplete((sendResult, exception) -> { + if (exception == null) { if (metadataChannel != null) { KafkaProducerMessageHandler.this.messagingTemplate.send(metadataChannel, - getMessageBuilderFactory().fromMessage(message) - .setHeader(KafkaHeaders.RECORD_METADATA, result.getRecordMetadata()).build()); + getMessageBuilderFactory() + .fromMessage(message) + .setHeader(KafkaHeaders.RECORD_METADATA, sendResult.getRecordMetadata()) + .build()); } } - - @Override - public void onFailure(Throwable ex) { - if (failureChannel != null) { - KafkaProducerMessageHandler.this.messagingTemplate.send(failureChannel, - KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage( - new KafkaSendFailureException(message, producerRecord, ex), null)); - } + else if (failureChannel != null) { + KafkaProducerMessageHandler.this.messagingTemplate.send(failureChannel, + KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage( + new KafkaSendFailureException(message, producerRecord, exception), null)); } - }); } @@ -720,51 +712,47 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes return new ConvertingReplyFuture(future); } - private final class ConvertingReplyFuture extends SettableListenableFuture { + private final class ConvertingReplyFuture extends CompletableFuture { ConvertingReplyFuture(RequestReplyFuture future) { addCallback(future); } private void addCallback(final RequestReplyFuture future) { - future.addCallback(new ListenableFutureCallback>() { // NOSONAR - - @Override - public void onSuccess(ConsumerRecord result) { + future.whenComplete((result, exception) -> { + if (exception == null) { try { - set(dontLeakHeaders(KafkaProducerMessageHandler.this.replyMessageConverter.toMessage(result, - null, null, KafkaProducerMessageHandler.this.replyPayloadType))); + complete(dontLeakHeaders( + KafkaProducerMessageHandler.this.replyMessageConverter.toMessage(result, null, null, + KafkaProducerMessageHandler.this.replyPayloadType))); } - catch (Exception e) { - setException(e); + catch (Exception ex) { + completeExceptionally(ex); } } - - private Message dontLeakHeaders(Message message) { - if (message.getHeaders() instanceof KafkaMessageHeaders) { - Map headers = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders(); - headers.remove(KafkaHeaders.CORRELATION_ID); - headers.remove(KafkaHeaders.REPLY_TOPIC); - headers.remove(KafkaHeaders.REPLY_PARTITION); - return message; - } - else { - return getMessageBuilderFactory().fromMessage(message) - .removeHeader(KafkaHeaders.CORRELATION_ID) - .removeHeader(KafkaHeaders.REPLY_TOPIC) - .removeHeader(KafkaHeaders.REPLY_PARTITION) - .build(); - } + else { + completeExceptionally(exception); } - - @Override - public void onFailure(Throwable ex) { - setException(ex); - } - }); } + private Message dontLeakHeaders(Message message) { + if (message.getHeaders() instanceof KafkaMessageHeaders) { + Map headers = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders(); + headers.remove(KafkaHeaders.CORRELATION_ID); + headers.remove(KafkaHeaders.REPLY_TOPIC); + headers.remove(KafkaHeaders.REPLY_PARTITION); + return message; + } + else { + return getMessageBuilderFactory().fromMessage(message) + .removeHeader(KafkaHeaders.CORRELATION_ID) + .removeHeader(KafkaHeaders.REPLY_TOPIC) + .removeHeader(KafkaHeaders.REPLY_PARTITION) + .build(); + } + } + } /** 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 f8ef5e7a40..75caaa55ce 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 @@ -39,6 +39,7 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -309,13 +310,14 @@ class KafkaProducerMessageHandlerTests { final RuntimeException fooException = new RuntimeException("foo"); - handler = new KafkaProducerMessageHandler<>(new KafkaTemplate(producerFactory) { + handler = new KafkaProducerMessageHandler<>(new KafkaTemplate<>(producerFactory) { @Override - protected ListenableFuture> doSend( + protected CompletableFuture> doSend( ProducerRecord producerRecord) { - SettableListenableFuture> future = new SettableListenableFuture<>(); - future.setException(fooException); + + CompletableFuture> future = new CompletableFuture<>(); + future.completeExceptionally(fooException); return future; }