From 8008ccaab95901dc200f31cedcd04e656ae99209 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 11 May 2020 20:40:48 +0200 Subject: [PATCH] Set 'setRequiredGroups' in StreamBridge This is set b y default for trully dynamic cases which are cases where source bindings such as 'spring.cloud.stream.source' were not created --- .../springframework/cloud/stream/function/StreamBridge.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 9e0989c50..228710be1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -131,7 +131,9 @@ public final class StreamBridge implements SmartInitializingSingleton { @SuppressWarnings({ "unchecked", "unused" }) public boolean send(String bindingName, Object data, MimeType outputContentType) { SubscribableChannel messageChannel = this.channelCache.get(bindingName); + ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); if (messageChannel == null) { + producerProperties.setRequiredGroups(bindingName); FunctionRegistration> fr = new FunctionRegistration<>(v -> v, bindingName); this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); messageChannel = new DirectWithAttributesChannel(); @@ -139,8 +141,6 @@ public final class StreamBridge implements SmartInitializingSingleton { this.channelCache.put(bindingName, messageChannel); } - ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); - //producerProperties.setRequiredGroups(bindingName); Function functionToInvoke = this.functionCatalog.lookup(bindingName, outputContentType.toString()); if (producerProperties != null && producerProperties.isPartitioned()) { functionToInvoke = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) functionToInvoke, this.applicationContext, producerProperties);