GH-2799: Fix AnnGateProxyFB for empty errChannel

Fixes #2799

**Cherry-pick to 5.1.x & 5.0.x**

When `AnnotationGatewayProxyFactoryBean` is used for the one-way (`void`)
POJI method invocation and downstream processing is based on an
`ExecutorChannel`, the default `errorChannel` value from the annotation
is resolved to the empty string with is set to the target gateway proxy
and can not be resolved to the target bean eventually in the
`MessagePublishingErrorHandler` in case of exception

* Check `errorChannel` attribute for empty string and don't set it into
the `errorChannelName`.
This way the `MessagePublishingErrorHandler` will resolve to the global
`errorChannel` as expected.

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/gateway/AnnotationGatewayProxyFactoryBean.java

@@ -95,7 +95,9 @@ protected void onInit() {
 		setDefaultReplyChannelName(defaultReplyChannel);
 
 		String errorChannel = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("errorChannel"));
-		setErrorChannelName(errorChannel);
+		if (StringUtils.hasText(errorChannel)) {
+			setErrorChannelName(errorChannel);
+		}
 
 		String asyncExecutor = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("asyncExecutor"));
 		if (asyncExecutor == null || AnnotationConstants.NULL.equals(asyncExecutor)) {
@@ -141,11 +143,15 @@ else if (StringUtils.hasText(asyncExecutor)) {
 
 		String defaultRequestTimeout =
 				beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultRequestTimeout"));
-		setDefaultRequestTimeout(Long.parseLong(defaultRequestTimeout));
+		if (StringUtils.hasText(defaultRequestTimeout)) {
+			setDefaultRequestTimeout(Long.parseLong(defaultRequestTimeout));
+		}
 
 		String defaultReplyTimeout =
 				beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultReplyTimeout"));
-		setDefaultReplyTimeout(Long.parseLong(defaultReplyTimeout));
+		if (StringUtils.hasText(defaultReplyTimeout)) {
+			setDefaultReplyTimeout(Long.parseLong(defaultReplyTimeout));
+		}
 
 		super.onInit();
 	}

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -33,6 +33,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.aopalliance.aop.Advice;
@@ -52,6 +53,7 @@
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Scope;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.annotation.MessageEndpoint;
@@ -491,6 +493,26 @@ public void testPrototypeIsNotOverridden() {
 				this.flow2WithPrototypeHandlerConsumer.getHandler());
 	}
 
+	@Autowired
+	@Qualifier("globalErrorChannelResolutionFunction")
+	private Consumer<String> globalErrorChannelResolutionGateway;
+
+	@Autowired
+	SubscribableChannel errorChannel;
+
+	@Test
+	public void testGlobalErrorChannelResolutionFlow() throws InterruptedException {
+		CountDownLatch errorMessageLatch = new CountDownLatch(1);
+		MessageHandler errorMessageHandler = m -> errorMessageLatch.countDown();
+		this.errorChannel.subscribe(errorMessageHandler);
+
+		this.globalErrorChannelResolutionGateway.accept("foo");
+
+		assertThat(errorMessageLatch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		this.errorChannel.unsubscribe(errorMessageHandler);
+	}
+
 	@MessagingGateway
 	public interface ControlBusGateway {
 
@@ -882,6 +904,16 @@ public IntegrationFlow flow2WithPrototypeHandler(
 			return f -> f.handle(handler, e -> e.id("flow2WithPrototypeHandlerConsumer"));
 		}
 
+		@Bean
+		public IntegrationFlow globalErrorChannelResolutionFlow(@Qualifier("taskScheduler") TaskExecutor taskExecutor) {
+			return IntegrationFlows.from(Consumer.class, "globalErrorChannelResolutionFunction")
+					.channel(c -> c.executor(taskExecutor))
+					.handle((GenericHandler<?>) (p, h) -> {
+						throw new RuntimeException("intentional");
+					})
+					.get();
+		}
+
 	}
 
 	@Service