RESOLVED - issue BATCH-722: No framework logic should depend on the value of ExitStatus

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

@@ -14,7 +14,6 @@ import org.springframework.batch.item.FlushFailedException;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.item.database.HibernateAwareItemWriter;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.integration.channel.MessageChannel;
@@ -30,7 +29,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport
 	/**
 	 * Key for items processed in the current transaction {@link RepeatContext}.
 	 */
-	private static final String ITEMS_PROCESSED = HibernateAwareItemWriter.class.getName() + ".ITEMS_PROCESSED";
+	private static final String ITEMS_PROCESSED = ChunkMessageChannelItemWriter.class.getName() + ".ITEMS_PROCESSED";
 
 	static final String ACTUAL = "ACTUAL";
 
@@ -47,6 +46,15 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport
 
 	private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
 
+	/**
+	 * Public setter for the throttle limit. This limits the number of pending
+	 * requests for chunk processing to avoid overwhelming the receivers.
+	 * @param throttleLimit the throttle limit to set
+	 */
+	public void setThrottleLimit(long throttleLimit) {
+		this.throttleLimit = throttleLimit;
+	}
+
 	public void setReplyChannel(MessageChannel replyChannel) {
 		this.replyChannel = replyChannel;
 	}
@@ -62,7 +70,10 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport
 	}
 
 	/**
-	 * Flush the buffer.
+	 * Flush the buffer, sending the items as a chunk message to be processed by
+	 * a {@link ChunkHandler}. To avoid overwhelming the receivers, this method
+	 * will block until the number of chunks pending is less than the throttle
+	 * limit.
 	 * 
 	 * @see org.springframework.batch.item.ItemWriter#flush()
 	 */
@@ -71,7 +82,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport
 		bindTransactionResources(); // in case we are called outside a
 		// transaction
 
-		// Block until expecting < throttle limit - can Spring
+		// Block until expecting <= throttle limit - can Spring
 		// Integration do that for me?
 		while (localState.getExpecting() > throttleLimit) {
 			getNextResult(100);
@@ -83,7 +94,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport
 
 			logger.debug("Dispatching chunk: " + processed);
 			ChunkRequest request = new ChunkRequest(processed, localState.getJobId(), localState.getSkipCount());
-			GenericMessage<ChunkRequest> message = new GenericMessage<ChunkRequest>(request );
+			GenericMessage<ChunkRequest> message = new GenericMessage<ChunkRequest>(request);
 			requestChannel.send(message);
 			localState.expected++;
 
@@ -95,7 +106,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport
 		unbindTransactionResources();
 
 	}
-	
+
 	@Override
 	public void beforeStep(StepExecution stepExecution) {
 		localState.setStepExecution(stepExecution);
@@ -169,8 +180,9 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport
 		if (message != null) {
 			ChunkResponse payload = (ChunkResponse) message.getPayload();
 			Long jobInstanceId = payload.getJobId();
-			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()+"].");
+			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++;
 			ExitStatus result = payload.getExitStatus();
 			// TODO: check it can never be ExitStatus.FINISHED?

spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java

@@ -24,7 +24,6 @@ import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.item.ExecutionContext;
-import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.beans.factory.annotation.Required;
 import org.springframework.integration.annotation.Handler;
 
@@ -62,7 +61,6 @@ public class StepExecutionMessageHandler {
 	@Handler
 	public JobExecutionRequest handle(JobExecutionRequest request) {
 
-
 		// Hand off immediately if the job has already failed
 		if (isComplete(request)) {
 			return request;
@@ -87,15 +85,23 @@ public class StepExecutionMessageHandler {
 			// skipped
 			if (shouldStart(lastStepExecution, step)) {
 
-				boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0 && !lastStepExecution
-						.getExitStatus().equals(ExitStatus.FINISHED)) ? true : false;
-
-				if (!isRestart || lastStepExecution == null) {
+				if (!isRestart(jobInstance, lastStepExecution)) {
 					stepExecution.setExecutionContext(new ExecutionContext());
 				}
 
 				step.execute(stepExecution);
 
+			}
+			else if (lastStepExecution != null) {
+
+				/*
+				 * We only set these if the step is not going to execute. They
+				 * might be needed by the next step to receive the request, but
+				 * they won't be persisted because the step is not executed.
+				 */
+				stepExecution.setStatus(lastStepExecution.getStatus());
+				stepExecution.setExitStatus(lastStepExecution.getExitStatus());
+
 			}
 
 			// (the job might actually not be complete, but the stage is).
@@ -114,6 +120,15 @@ public class StepExecutionMessageHandler {
 
 	}
 
+	/**
+	 * @param jobInstance
+	 * @param lastStepExecution
+	 * @return
+	 */
+	private boolean isRestart(JobInstance jobInstance, StepExecution lastStepExecution) {
+		return (lastStepExecution != null && !lastStepExecution.getStatus().equals(BatchStatus.COMPLETED));
+	}
+
 	/**
 	 * @param request
 	 * @return
@@ -151,7 +166,7 @@ public class StepExecutionMessageHandler {
 
 		if (stepStatus == BatchStatus.UNKNOWN) {
 			throw new JobExecutionException("Cannot restart step from UNKNOWN status.  "
-					+ "The last execution ended with a failure that could not be rolled back, "
+					+ "The last execution may have ended with a failure that could not be rolled back, "
 					+ "so it may be dangerous to proceed.  " + "Manual intervention is probably necessary.");
 		}