From f58dcf371db7cadb9ddf9de9ef018fa575ec10fd Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 25 Mar 2022 14:26:25 +0100 Subject: [PATCH] Fix concurrency issue in StreamBridge Resolves #2300 --- .../cloud/stream/function/StreamBridge.java | 20 ++++++++++++------- .../binder/tck/ContentTypeTckTests.java | 5 +++-- 2 files changed, 16 insertions(+), 9 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 57baec1a3..e9af62520 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -208,20 +208,26 @@ public final class StreamBridge implements SmartInitializingSingleton { @SuppressWarnings({ "unchecked"}) public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) { - if (data instanceof Message) { - data = MessageBuilder.fromMessage((Message) data).setHeader(MessageUtils.TARGET_PROTOCOL, "streamBridge").build(); - } - else { - data = new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, "streamBridge")); - } + + ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName); Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties); + if (producerProperties != null && producerProperties.isPartitioned()) { functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties); } - Message resultMessage = (Message) functionToInvoke.apply(data); + + + Message messageToSend = data instanceof Message + ? MessageBuilder.fromMessage((Message) data).setHeader(MessageUtils.TARGET_PROTOCOL, "streamBridge").build() + : new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, "streamBridge")); + + Message resultMessage; + synchronized (this) { + resultMessage = (Message) functionToInvoke.apply(messageToSend); + } return messageChannel.send(resultMessage); } diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java index bf42fbe43..823756368 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java @@ -267,10 +267,11 @@ public class ContentTypeTckTests { .setHeader("contentType", new MimeType("text", "plain")).build()); Message outputMessage = target.receive(); + System.out.println(new String(outputMessage.getPayload())); assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)) .isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE); - assertThat(new String(outputMessage.getPayload(), StandardCharsets.UTF_8)) - .isEqualTo(jsonPayload); + assertThat(outputMessage.getPayload()) + .isEqualTo(jsonPayload.getBytes()); } @Test