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

GH-2610 Fix default error handling

Browse Source 
Ensures that default error handling is preserved if user did not define binder specific error handler

Resolves #2610
This commit is contained in:
Oleg Zhurakousky
2023-01-05 15:02:29 +01:00
parent 8fb166fc58
commit 686a75c077
5 changed files with 35 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,6 +25,7 @@ import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.test.InputDestination;
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;
@@ -80,6 +81,22 @@ public class ErrorBindingTests {
assertThat(errorConfiguration.counter).isEqualTo(6);
}
@Test
void testConfigurationWithoutBinderSpecificErrorHandler() {
ApplicationContext context = 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.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
source.send(new GenericMessage<byte[]>("Hello".getBytes()));
TestChannelBinder binder = context.getBean(TestChannelBinder.class);
assertThat(binder.getLastError()).isNotNull();
}
@EnableAutoConfiguration
public static class TestProcessor {
@@ -89,6 +106,17 @@ public class ErrorBindingTests {
}
}
@EnableAutoConfiguration
public static class NoErrorHandler {
@Bean
public Function<String, String> process() {
return s -> {
throw new RuntimeException("intentional");
};
}
}
@EnableAutoConfiguration
public static class ErrorConfigurationDefault {

106
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ErrorHandlingTests.java
View File

@@ -1,106 +0,0 @@
/*
* Copyright 2019-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.tck;
import java.util.function.Function;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Oleg Zhurakousky
*
*/
@Disabled
public class ErrorHandlingTests {
@Test
void testGlobalErrorWithMessage() {
ApplicationContext context = new SpringApplicationBuilder(GlobalErrorHandlerWithErrorMessageConfig.class)
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
source.send(new GenericMessage<>("foo".getBytes()));
GlobalErrorHandlerWithErrorMessageConfig config = context
.getBean(GlobalErrorHandlerWithErrorMessageConfig.class);
assertThat(config.globalErroInvoked).isTrue();
}
@Test
void testGlobalErrorWithThrowable() {
ApplicationContext context = new SpringApplicationBuilder(GlobalErrorHandlerWithThrowableConfig.class)
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
source.send(new GenericMessage<>("foo".getBytes()));
GlobalErrorHandlerWithThrowableConfig config = context.getBean(GlobalErrorHandlerWithThrowableConfig.class);
assertThat(config.globalErroInvoked).isTrue();
}
@Import(TestChannelBinderConfiguration.class)
@EnableAutoConfiguration
public static class GlobalErrorHandlerWithErrorMessageConfig {
private boolean globalErroInvoked;
@Bean
public Function<String, String> func() {
return v -> {
throw new RuntimeException("test exception");
};
}
@ServiceActivator(inputChannel = "errorChannel")
public void generalError(Message<?> message) {
this.globalErroInvoked = true;
}
}
@Import(TestChannelBinderConfiguration.class)
@EnableAutoConfiguration
public static class GlobalErrorHandlerWithThrowableConfig {
private boolean globalErroInvoked;
@Bean
public Function<String, String> func() {
return v -> {
throw new RuntimeException("test exception");
};
}
@ServiceActivator(inputChannel = "errorChannel")
public void generalError(Throwable exception) {
this.globalErroInvoked = true;
}
}
}

2
core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java
View File

@@ -190,7 +190,7 @@ public class TestChannelBinder extends
protected MessageHandler getErrorMessageHandler(ConsumerDestination destination,
String group, ConsumerProperties consumerProperties) {
return m -> {
this.logger.debug("Error handled: " + m);
this.logger.info("Error handled: " + m);
this.lastError = m;
};
}

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

@@ -774,7 +774,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
errorChannelName, SubscribableChannel.class, () -> binderErrorChannel);
this.subscribeFunctionErrorHandler(errorChannelName, consumerProperties.getBindingName());
//
ErrorMessageSendingRecoverer recoverer = new ErrorMessageSendingRecoverer(binderErrorChannel, errorMessageStrategy);
String recovererBeanName = getErrorRecovererName(destination, group, consumerProperties);
if (!getApplicationContext().containsBean(recovererBeanName)) {
@@ -789,13 +788,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
String errorMessageHandlerName = getErrorMessageHandlerName(destination, group,
consumerProperties);
if (binderProvidedErrorHandler == null) {
binderProvidedErrorHandler = this.getDefaultErrorMessageHandler(binderErrorChannel, polled);
}
if (binderProvidedErrorHandler != null) {
if (this.isSubscribable(binderErrorChannel)) {
if (!getApplicationContext().containsBean(errorMessageHandlerName)) {
MessageHandler h = binderProvidedErrorHandler;
((GenericApplicationContext) getApplicationContext()).registerBean(
errorMessageHandlerName, MessageHandler.class,
() -> h);
((GenericApplicationContext) getApplicationContext()).registerBean(errorMessageHandlerName, MessageHandler.class, () -> h);
binderErrorChannel.subscribe(binderProvidedErrorHandler);
}
else {
@@ -930,9 +930,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
* @return the handler.
*/
protected MessageHandler getDefaultErrorMessageHandler(
LastSubscriberAwareChannel errorChannel, boolean defaultErrorChannelPresent) {
return new FinalRethrowingErrorMessageHandler(errorChannel,
defaultErrorChannelPresent);
SubscribableChannel errorChannel, boolean defaultErrorChannelPresent) {
return new FinalRethrowingErrorMessageHandler();
}
/**

15
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java
View File

@@ -33,23 +33,8 @@ import org.springframework.messaging.MessagingException;
class FinalRethrowingErrorMessageHandler
implements MessageHandler, LastSubscriberMessageHandler {
private final LastSubscriberAwareChannel errorChannel;
private final boolean defaultErrorChannelPresent;
FinalRethrowingErrorMessageHandler(LastSubscriberAwareChannel errorChannel,
boolean defaultErrorChannelPresent) {
this.errorChannel = errorChannel;
this.defaultErrorChannelPresent = defaultErrorChannelPresent;
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (this.errorChannel.subscribers() > (this.defaultErrorChannelPresent ? 2 : 1)) {
// user has subscribed; default is 2, this and the bridge to the
// errorChannel
return;
}
if (message.getPayload() instanceof MessagingException) {
throw (MessagingException) message.getPayload();
}