From 8223c8d3f7874ffddd2335b9491c6745cf817dc9 Mon Sep 17 00:00:00 2001 From: abilan Date: Wed, 5 Apr 2023 17:14:38 -0400 Subject: [PATCH] Fix producer for outputChannel Current async version of the `KinesisMessageHandler` is request-reply. The binder deals with it as a one-way, so no need in reply. * Add `messageHandler.setOutputChannel(new NullChannel())` for the producer --- .../stream/binder/kinesis/KinesisMessageChannelBinder.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java index a161132..a77e60d 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java @@ -61,6 +61,7 @@ import org.springframework.integration.aws.outbound.AbstractAwsMessageHandler; import org.springframework.integration.aws.outbound.KinesisMessageHandler; import org.springframework.integration.aws.outbound.KplMessageHandler; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.expression.ExpressionUtils; @@ -230,6 +231,7 @@ protected MessageHandler createProducerMessageHandler(ProducerDestination destin messageHandler.setAsync(!producerProperties.getExtension().isSync()); messageHandler.setSendTimeout(producerProperties.getExtension().getSendTimeout()); messageHandler.setBeanFactory(getBeanFactory()); + messageHandler.setOutputChannel(new NullChannel()); if (errorChannel != null) { ((InterceptableChannel) channel)