SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2599 Fix error binding when mapped to multiple destinations

Browse Source 
Resolves #2599
This commit is contained in:
Oleg Zhurakousky
2023-01-05 15:18:45 +01:00
parent 686a75c077
commit a9196584d2
2 changed files with 17 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java
View File

@@ -97,6 +97,19 @@ public class ErrorBindingTests {
assertThat(binder.getLastError()).isNotNull();
}
@Test
void testErrorBindingWithMultipleDestinationPerBinding() {
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(NoErrorHandler.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.bindings.process-in-0.consumer.max-attempts=1",
"--spring.cloud.function.definition=process",
"--spring.cloud.stream.bindings.process-in-0.destination=one,two",
"--spring.jmx.enabled=false");
// must not fail GH-2599
}
@EnableAutoConfiguration
public static class TestProcessor {

7
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -741,7 +741,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
boolean polled) {
ErrorMessageStrategy errorMessageStrategy = getErrorMessageStrategy();
String errorChannelName = errorsBaseName(destination, group, consumerProperties);
BindingServiceProperties bsp = this.getBindingServiceProperties();
FunctionInvocationWrapper userErrorHandler = null;
@@ -770,8 +769,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
binderErrorChannel = new BinderErrorChannel();
}
binderErrorChannel.setComponentName(errorChannelName);
((GenericApplicationContext) getApplicationContext()).registerBean(
errorChannelName, SubscribableChannel.class, () -> binderErrorChannel);
if (!this.getApplicationContext().containsBean(errorChannelName)) {
((GenericApplicationContext) getApplicationContext()).registerBean(
errorChannelName, SubscribableChannel.class, () -> binderErrorChannel);
}
this.subscribeFunctionErrorHandler(errorChannelName, consumerProperties.getBindingName());
ErrorMessageSendingRecoverer recoverer = new ErrorMessageSendingRecoverer(binderErrorChannel, errorMessageStrategy);