INT-3781: Configure HTTP GW Timeout Status Code

JIRA: https://jira.spring.io/browse/INT-3781

- Add mechanisms to set status code on an inbound gateway timeout.
- Send a message to the error channel if configured.

Polishing - Add failedMessage to MTE

Fix Schema Docs; Timeout Detection on Error Flow

Add Zookeeper Leadership Logs

Fix typo in the `SmartLifecycleRoleController`

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.gateway;
 
+import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
@@ -50,6 +51,16 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 
 	private static final long DEFAULT_TIMEOUT = 1000L;
 
+	private final SimpleMessageConverter messageConverter = new SimpleMessageConverter();
+
+	private final MessagingTemplate messagingTemplate;
+
+	private final HistoryWritingMessagePostProcessor historyWritingPostProcessor =
+			new HistoryWritingMessagePostProcessor();
+
+	private final Object replyMessageCorrelatorMonitor = new Object();
+
+	private final boolean errorOnTimeout;
 
 	private volatile MessageChannel requestChannel;
 
@@ -68,26 +79,34 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 	@SuppressWarnings("rawtypes")
 	private volatile InboundMessageMapper requestMapper = new DefaultRequestMapper();
 
-	private final SimpleMessageConverter messageConverter = new SimpleMessageConverter();
-
-	private final MessagingTemplate messagingTemplate;
-
-	private final HistoryWritingMessagePostProcessor historyWritingPostProcessor =
-			new HistoryWritingMessagePostProcessor();
-
 	private volatile boolean initialized;
 
 	private volatile AbstractEndpoint replyMessageCorrelator;
 
-	private final Object replyMessageCorrelatorMonitor = new Object();
-
 
+	/**
+	 * Construct an instance that will return null if no reply is received.
+	 */
 	public MessagingGatewaySupport() {
+		this(false);
+	}
+
+	/**
+	 * If errorOnTimeout is true, construct an instance that will send an
+	 * {@link ErrorMessage} with a {@link MessageTimeoutException} payload to the error
+	 * channel if a reply is expected but none is received. If no error channel is
+	 * configured, the {@link MessageTimeoutException} will be thrown.
+	 *
+	 * @param errorOnTimeout true to create the error message.
+	 * @since 4.2
+	 */
+	public MessagingGatewaySupport(boolean errorOnTimeout) {
 		MessagingTemplate template = new MessagingTemplate();
 		template.setMessageConverter(this.messageConverter);
 		template.setSendTimeout(DEFAULT_TIMEOUT);
 		template.setReceiveTimeout(this.replyTimeout);
 		this.messagingTemplate = template;
+		this.errorOnTimeout = errorOnTimeout;
 	}
 
 
@@ -333,6 +352,14 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 					error = ((ErrorMessage) reply).getPayload();
 				}
 			}
+			if (reply == null && this.errorOnTimeout) {
+				if (object instanceof Message) {
+					error = new MessageTimeoutException((Message<?>) object, "No reply received within timeout");
+				}
+				else {
+					error = new MessageTimeoutException("No reply received within timeout");
+				}
+			}
 		}
 		catch (Exception e) {
 			if (logger.isDebugEnabled()) {
@@ -363,6 +390,15 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
 				if (errorFlowReply != null && errorFlowReply.getPayload() instanceof Throwable) {
 					this.rethrow((Throwable) errorFlowReply.getPayload(), "error flow returned an Error Message");
 				}
+				if (errorFlowReply == null && this.errorOnTimeout) {
+					if (object instanceof Message) {
+						throw new MessageTimeoutException((Message<?>) object,
+								"No reply received from error channel within timeout");
+					}
+					else {
+						throw new MessageTimeoutException("No reply received from error channel within timeout");
+					}
+				}
 				return errorFlowReply;
 			}
 			else { // no errorChannel so we'll propagate

spring-integration-core/src/main/java/org/springframework/integration/support/SmartLifecycleRoleController.java

@@ -141,15 +141,23 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
 				}
 
 			});
+			if (logger.isDebugEnabled()) {
+				logger.debug("Zookeeper leadership granted: Starting: " + lifecycles);
+			}
 			for (SmartLifecycle lifecycle : lifecycles) {
 				try {
 					lifecycle.start();
 				}
 				catch (Exception e) {
-					logger.error("Failed to start " + lifecycle + " in role " + role);
+					logger.error("Failed to start " + lifecycle + " in role " + role, e);
 				}
 			}
 		}
+		else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Zookeeper leadership granted: Nothing to do");
+			}
+		}
 	}
 
 	/**
@@ -172,15 +180,23 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
 				}
 
 			});
+			if (logger.isDebugEnabled()) {
+				logger.debug("Zookeeper leadership revoked: Stopping: " + lifecycles);
+			}
 			for (SmartLifecycle lifecycle : lifecycles) {
 				try {
 					lifecycle.stop();
 				}
 				catch (Exception e) {
-					logger.error("Failed to stop " + lifecycle + " in role " + role);
+					logger.error("Failed to stop " + lifecycle + " in role " + role, e);
 				}
 			}
 		}
+		else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Zookeeper leadership revoked: Nothing to do");
+			}
+		}
 	}
 
 	private void addLazyLifecycles() {