BATCHADM-49: Added redelivered flag to ChunkResponse

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java

@@ -95,7 +95,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 			ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState
 					.createStepContribution());
 			messagingGateway.send(request);
-			localState.expected.incrementAndGet();
+			localState.incrementExpected();
 
 		}
 
@@ -201,8 +201,17 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 			Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
 			Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["
 					+ jobInstanceId + "] should have been [" + localState.getJobId() + "].");
-			localState.actual.incrementAndGet();
+			if (payload.isRedelivered()) {
+				logger
+						.warn("Redelivered result detected, which may indicate stale state. In the best case, we just picked up a timed out message "
+								+ "from a previous failed execution. In the worst case (and if this is not a restart), "
+								+ "the step may now timeout.  In that case if you believe that all messages "
+								+ "from workers have been sent, the business state "
+								+ "is probably inconsistent, and the step will fail.");
+				localState.incrementRedelivered();
+			}
 			localState.pushStepContribution(payload.getStepContribution());
+			localState.incrementActual();
 			if (!payload.isSuccessful()) {
 				throw new AsynchronousFailureException("Failure or interrupt detected in handler: "
 						+ payload.getMessage());
@@ -232,6 +241,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 		private AtomicInteger expected = new AtomicInteger();
 
+		private AtomicInteger redelivered = new AtomicInteger();
+
 		private StepExecution stepExecution;
 
 		private Queue<StepContribution> contributions = new LinkedBlockingQueue<StepContribution>();
@@ -263,6 +274,18 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 			}
 		}
 
+		public void incrementRedelivered() {
+			redelivered.incrementAndGet();
+		}
+
+		public void incrementActual() {
+			actual.incrementAndGet();
+		}
+
+		public void incrementExpected() {
+			expected.incrementAndGet();
+		}
+
 		public StepContribution createStepContribution() {
 			return stepExecution.createStepContribution();
 		}

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java

@@ -35,6 +35,8 @@ public class ChunkResponse implements Serializable {
 	private final boolean status;
 
 	private final String message;
+	
+	private final boolean redelivered;
 
 	public ChunkResponse(Long jobId, StepContribution stepContribution) {
 		this(true, jobId, stepContribution, null);
@@ -45,10 +47,19 @@ public class ChunkResponse implements Serializable {
 	}
 
 	public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message) {
+		this(status, jobId, stepContribution, message, false);
+	}
+
+	public ChunkResponse(ChunkResponse input, boolean redelivered) {
+		this(input.status, input.jobId, input.stepContribution, input.message, redelivered);
+	}
+
+	public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message, boolean redelivered) {
 		this.status = status;
 		this.jobId = jobId;
 		this.stepContribution = stepContribution;
 		this.message = message;
+		this.redelivered = redelivered;
 	}
 
 	public StepContribution getStepContribution() {
@@ -62,6 +73,10 @@ public class ChunkResponse implements Serializable {
 	public boolean isSuccessful() {
 		return status;
 	}
+	
+	public boolean isRedelivered() {
+		return redelivered;
+	}
 
 	public String getMessage() {
 		return message;

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/JmsRedeliveredExtractor.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2009-2010 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.batch.integration.chunk;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.annotation.Header;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class JmsRedeliveredExtractor {
+	
+	private static final Log logger = LogFactory.getLog(JmsRedeliveredExtractor.class);
+	
+	public ChunkResponse extract(ChunkResponse input, @Header("springintegration_jms_redelivered") boolean redelivered) {
+		logger.debug("Extracted redelivered flag for response, value="+redelivered);
+		return new ChunkResponse(input, redelivered);
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java

@@ -26,6 +26,8 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im
 
 	private MessageSource<?> source;
 
+	private MessageChannel channel;
+
 	/**
 	 * Convenient default constructor for configuration purposes.
 	 */
@@ -39,6 +41,16 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im
 		this.source = source;
 	}
 
+	/**
+	 * Optional MessageChannel for injecting the message receieved from the source (defaults to the channel
+	 * intercepted in {@link #preReceive(MessageChannel)}).
+	 * 
+	 * @param channel the channel to set
+	 */
+	public void setChannel(MessageChannel channel) {
+		this.channel = channel;
+	}
+
 	/**
 	 * Asserts that mandatory properties are set.
 	 * @see InitializingBean#afterPropertiesSet()
@@ -64,6 +76,9 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im
 	public boolean preReceive(MessageChannel channel) {
 		Message<?> message = source.receive();
 		if (message != null) {
+			if (this.channel!=null) {
+				channel = this.channel;
+			}
 			channel.send(message);
 			if (logger.isDebugEnabled()) {
 				logger.debug("Sent " + message + " to channel " + channel.getName());