INT-4527: Delayer Retries and Error Handling

JIRA: https://jira.spring.io/browse/INT-4527
Fixes #2543

- Add error handling within the scope of a transaction
- Add `retryDelay` and `maxAttempts`
- Include the delivery attempt header in the `ErrorMessage`
- Add `transactionalRelease()` methods to the DSL

Check context in `ContextRefreshedEvent`.

Complete test case; don't schedule for re-release after a successful send after a failure.

Use `identityHashCode` for deliveries map key - error flow might change the message `hashCode`.

Debug logging

Polishing; remove parameter from the `DelayerEndpointSpec.transactionalRelease()`

* Polishing some code style

Author: garyrussell

spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java

@@ -67,7 +67,7 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
 
 	public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback";
 
-	private Set<String> readOnlyHeaders = new HashSet<String>();
+	private Set<String> readOnlyHeaders = new HashSet<>();
 
 	public IntegrationMessageHeaderAccessor(Message<?> message) {
 		super(message);
@@ -84,33 +84,33 @@ public IntegrationMessageHeaderAccessor(Message<?> message) {
 	public void setReadOnlyHeaders(String... readOnlyHeaders) {
 		Assert.noNullElements(readOnlyHeaders, "'readOnlyHeaders' must not be contain null items.");
 		if (!ObjectUtils.isEmpty(readOnlyHeaders)) {
-			this.readOnlyHeaders = new HashSet<String>(Arrays.asList(readOnlyHeaders));
+			this.readOnlyHeaders = new HashSet<>(Arrays.asList(readOnlyHeaders));
 		}
 	}
 
 	@Nullable
 	public Long getExpirationDate() {
-		return this.getHeader(EXPIRATION_DATE, Long.class);
+		return getHeader(EXPIRATION_DATE, Long.class);
 	}
 
 	@Nullable
 	public Object getCorrelationId() {
-		return this.getHeader(CORRELATION_ID);
+		return getHeader(CORRELATION_ID);
 	}
 
 	public int getSequenceNumber() {
-		Number sequenceNumber = this.getHeader(SEQUENCE_NUMBER, Number.class);
+		Number sequenceNumber = getHeader(SEQUENCE_NUMBER, Number.class);
 		return (sequenceNumber != null ? sequenceNumber.intValue() : 0);
 	}
 
 	public int getSequenceSize() {
-		Number sequenceSize = this.getHeader(SEQUENCE_SIZE, Number.class);
+		Number sequenceSize = getHeader(SEQUENCE_SIZE, Number.class);
 		return (sequenceSize != null ? sequenceSize.intValue() : 0);
 	}
 
 	@Nullable
 	public Integer getPriority() {
-		Number priority = this.getHeader(PRIORITY, Number.class);
+		Number priority = getHeader(PRIORITY, Number.class);
 		return (priority != null ? priority.intValue() : null);
 	}
 
@@ -133,19 +133,20 @@ public Closeable getCloseableResource() {
 	 * @return the callback.
 	 * @since 5.0.1
 	 */
+	@Nullable
 	public AcknowledgmentCallback getAcknowledgmentCallback() {
 		return getHeader(ACKNOWLEDGMENT_CALLBACK, AcknowledgmentCallback.class);
 	}
 
 	/**
-	 * When a message-driven enpoint supports retry implicitly, this
+	 * When a message-driven endpoint supports retry implicitly, this
 	 * header is incremented for each delivery attempt.
 	 * @return the delivery attempt.
 	 * @since 5.0.1
 	 */
 	@Nullable
 	public AtomicInteger getDeliveryAttempt() {
-		return this.getHeader(DELIVERY_ATTEMPT, AtomicInteger.class);
+		return getHeader(DELIVERY_ATTEMPT, AtomicInteger.class);
 	}
 
 	@SuppressWarnings("unchecked")

spring-integration-core/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -27,19 +27,27 @@
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.handler.DelayHandler;
 import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.transaction.TransactionInterceptorBuilder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
 import org.springframework.util.Assert;
 
 /**
  * A {@link ConsumerEndpointSpec} for a {@link DelayHandler}.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
 public final class DelayerEndpointSpec extends ConsumerEndpointSpec<DelayerEndpointSpec, DelayHandler> {
 
-	private final List<Advice> delayedAdvice = new LinkedList<Advice>();
+	private final List<Advice> delayedAdvice = new LinkedList<>();
 
 	DelayerEndpointSpec(DelayHandler delayHandler) {
 		super(delayHandler);
@@ -97,6 +105,105 @@ public DelayerEndpointSpec delayExpression(String delayExpression) {
 		return this;
 	}
 
+	/**
+	 * Set a message channel to which an {@link ErrorMessage} will be sent if sending the
+	 * released message fails. If the error flow returns normally, the release is complete.
+	 * If the error flow throws an exception, the release will be re-attempted.
+	 * If there is a transaction advice on the release task, the error flow is called
+	 * within the transaction.
+	 * @param channel the channel.
+	 * @return the endpoint spec.
+	 * @see #maxAttempts(int)
+	 * @see #retryDelay(long)
+	 * @since 5.0.8
+	 */
+	public DelayerEndpointSpec delayedMessageErrorChannel(MessageChannel channel) {
+		this.handler.setDelayedMessageErrorChannel(channel);
+		return this;
+	}
+
+	/**
+	 * Set a message channel name to which an {@link ErrorMessage} will be sent if sending
+	 * the released message fails. If the error flow returns normally, the release is
+	 * complete. If the error flow throws an exception, the release will be re-attempted.
+	 * If there is a transaction advice on the release task, the error flow is called
+	 * within the transaction.
+	 * @param channel the channel name.
+	 * @return the endpoint spec.
+	 * @see #maxAttempts(int)
+	 * @see #retryDelay(long)
+	 * @since 5.0.8
+	 */
+	public DelayerEndpointSpec delayedMessageErrorChannel(String channel) {
+		this.handler.setDelayedMessageErrorChannelName(channel);
+		return this;
+	}
+
+	/**
+	 * Set the maximum number of release attempts for when message release fails.
+	 * Default {@value DelayHandler#DEFAULT_MAX_ATTEMPTS}.
+	 * @param maxAttempts the max attempts.
+	 * @return the endpoint spec.
+	 * @see #retryDelay(long)
+	 * @since 5.0.8
+	 */
+	public DelayerEndpointSpec maxAttempts(int maxAttempts) {
+		this.handler.setMaxAttempts(maxAttempts);
+		return this;
+	}
+
+	/**
+	 * Set an additional delay to apply when retrying after a release failure.
+	 * Default {@value DelayHandler#DEFAULT_RETRY_DELAY}.
+	 * @param retryDelay the retry delay.
+	 * @return the endpoint spec.
+	 * @see #maxAttempts(int)
+	 * @since 5.0.8
+	 */
+	public DelayerEndpointSpec retryDelay(long retryDelay) {
+		this.handler.setRetryDelay(retryDelay);
+		return this;
+	}
+
+	/**
+	 * Specify a {@link TransactionInterceptor} {@link Advice} with default
+	 * {@link PlatformTransactionManager} and {@link DefaultTransactionAttribute} for the
+	 * {@link MessageHandler}.
+	 * @return the spec.
+	 * @since 5.0.8
+	 */
+	public DelayerEndpointSpec transactionalRelease() {
+		TransactionInterceptor transactionInterceptor = new TransactionInterceptorBuilder().build();
+		this.componentsToRegister.put(transactionInterceptor, null);
+		return delayedAdvice(transactionInterceptor);
+	}
+
+	/**
+	 * Specify a {@link TransactionInterceptor} {@link Advice} for the {@link MessageHandler}.
+	 * @param transactionInterceptor the {@link TransactionInterceptor} to use.
+	 * @return the spec.
+	 * @see TransactionInterceptorBuilder
+	 * @since 5.0.8
+	 */
+	public DelayerEndpointSpec transactionalRelease(TransactionInterceptor transactionInterceptor) {
+		return delayedAdvice(transactionInterceptor);
+	}
+
+	/**
+	 * Specify a {@link TransactionInterceptor} {@link Advice} with the provided
+	 * {@code PlatformTransactionManager} and default {@link DefaultTransactionAttribute}
+	 * for the {@link MessageHandler}.
+	 * @param transactionManager the {@link PlatformTransactionManager} to use.
+	 * @return the spec.
+	 * @since 5.0.8
+	 */
+	public DelayerEndpointSpec transactionalRelease(PlatformTransactionManager transactionManager) {
+		return transactionalRelease(
+				new TransactionInterceptorBuilder()
+						.transactionManager(transactionManager)
+						.build());
+	}
+
 	/**
 	 * Specify the function to determine delay value against {@link Message}.
 	 * Typically used with a Java 8 Lambda expression: