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

Merge pull request #2998 from hightea/fix-duplicate-errors-logs

Browse Source 
Avoid duplicate error logging when multiple destinations per binding
This commit is contained in:
Oleg Zhurakousky
2024-09-05 02:42:57 -07:00
committed by GitHub
parent 64dd18cc8e a64a6e53e1
commit f456405b02
2 changed files with 34 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder;
import java.util.function.Consumer;
import java.util.function.Function;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Test;
import org.springframework.boot.WebApplicationType;
@@ -29,6 +30,9 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinder;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
@@ -99,7 +103,7 @@ class ErrorBindingTests {
@Test
void errorBindingWithMultipleDestinationPerBinding() {
new SpringApplicationBuilder(
ApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(NoErrorHandler.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.bindings.process-in-0.consumer.max-attempts=1",
@@ -108,6 +112,27 @@ class ErrorBindingTests {
"--spring.jmx.enabled=false");
// must not fail GH-2599
InputDestination source = context.getBean(InputDestination.class);
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
// We validate that error is logged only once : BridgeHandler to bean 'errorChannel' subscribed only once
BinderErrorChannel binderErrorChannel = context.getBean(BinderErrorChannel.class);
assertThat(binderErrorChannel.getSubscriberCount())
.isEqualTo(2); // binderProvidedErrorHandler and BridgeHandler to bean 'errorChannel'
// The BridgeHandler bean associated with this binding bridges to 'errorChannel' bean
assertThat(
context.getBeansOfType(BridgeHandler.class)
.entrySet()
.stream()
.filter(entry -> entry.getKey().endsWith("process-in-0.errors.bridge"))
.findAny())
.isPresent()
.get()
.satisfies(bridgeHandler -> assertThat(bridgeHandler.getValue().getOutputChannel())
.isNotNull()
.asInstanceOf(InstanceOfAssertFactories.type(AbstractMessageChannel.class))
.extracting(AbstractMessageChannel::getBeanName)
.isEqualTo(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
);
}
@EnableAutoConfiguration

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

@@ -886,10 +886,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
// Setup a bridge to global errorChannel to ensure logging of errors could be controlled via standard SI way
if (this.getApplicationContext().containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) && this.isSubscribable(binderErrorChannel)) {
SubscribableChannel globalErrorChannel = this.getApplicationContext().getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class);
BridgeHandler bridge = new BridgeHandler();
bridge.setOutputChannel(globalErrorChannel);
binderErrorChannel.subscribe(bridge);
String errorBridgeHandlerName = getErrorBridgeName(destination, group, consumerProperties);
if (!getApplicationContext().containsBean(errorBridgeHandlerName)) {
SubscribableChannel globalErrorChannel = this.getApplicationContext().getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class);
BridgeHandler bridge = new BridgeHandler();
bridge.setOutputChannel(globalErrorChannel);
binderErrorChannel.subscribe(bridge);
((GenericApplicationContext) getApplicationContext()).registerBean(errorBridgeHandlerName, BridgeHandler.class, () -> bridge);
}
}
return new ErrorInfrastructure(binderErrorChannel, recoverer, binderProvidedErrorHandler);
}