GH-1618 Fixed regression for global error channel

Fixed regression in global erro channel conversion caused by introduction of custom argument resolvers
Added tests

Resoilves #1618

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartMessageMethodArgumentResolver.java

@@ -27,6 +27,7 @@ import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver;
 import org.springframework.messaging.handler.annotation.support.MethodArgumentTypeMismatchException;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
@@ -68,7 +69,8 @@ class SmartMessageMethodArgumentResolver extends MessageMethodArgumentResolver {
 
 		Class<?> payloadClass = message.getPayload().getClass();
 
-		if (conversionNotRequired(payloadClass, targetPayloadType)) {
+		if (message instanceof ErrorMessage
+				|| conversionNotRequired(payloadClass, targetPayloadType)) {
 			return message;
 		}
 		Object payload = message.getPayload();

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ErrorHandlingTests.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2019-2019 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
+ *
+ *      http://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 org.junit.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+public class ErrorHandlingTests {
+
+	@Test
+	public 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
+	public 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();
+	}
+
+	@EnableBinding(Processor.class)
+	@Import(TestChannelBinderConfiguration.class)
+	@EnableAutoConfiguration
+	public static class GlobalErrorHandlerWithErrorMessageConfig {
+
+		private boolean globalErroInvoked;
+
+		@StreamListener(target = Processor.INPUT)
+		public void input(final String value) {
+			throw new RuntimeException("test exception");
+		}
+
+		@StreamListener("errorChannel")
+		public void generalError(Message<?> message) {
+			this.globalErroInvoked = true;
+		}
+
+	}
+
+	@EnableBinding(Processor.class)
+	@Import(TestChannelBinderConfiguration.class)
+	@EnableAutoConfiguration
+	public static class GlobalErrorHandlerWithThrowableConfig {
+
+		private boolean globalErroInvoked;
+
+		@StreamListener(target = Processor.INPUT)
+		public void input(final String value) {
+			throw new RuntimeException("test exception");
+		}
+
+		@StreamListener("errorChannel")
+		public void generalError(Throwable exception) {
+			this.globalErroInvoked = true;
+		}
+
+	}
+
+}