diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java index d8653982d..c1d394c1b 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java @@ -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("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 process() { + return s -> { + throw new RuntimeException("intentional"); + }; + } + } + @EnableAutoConfiguration public static class ErrorConfigurationDefault { diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ErrorHandlingTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ErrorHandlingTests.java deleted file mode 100644 index 565ea34ba..000000000 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ErrorHandlingTests.java +++ /dev/null @@ -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 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 func() { - return v -> { - throw new RuntimeException("test exception"); - }; - } - - @ServiceActivator(inputChannel = "errorChannel") - public void generalError(Throwable exception) { - this.globalErroInvoked = true; - } - - } - -} diff --git a/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java b/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java index e4db1f95d..6d87e0225 100644 --- a/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java +++ b/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java @@ -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; }; } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 014022c50..f00fa6113 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -774,7 +774,6 @@ public abstract class AbstractMessageChannelBinder 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 h); + ((GenericApplicationContext) getApplicationContext()).registerBean(errorMessageHandlerName, MessageHandler.class, () -> h); binderErrorChannel.subscribe(binderProvidedErrorHandler); } else { @@ -930,9 +930,8 @@ public abstract class AbstractMessageChannelBinder 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(); }