diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java index adb00b8f9..41b8193a9 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java @@ -21,14 +21,14 @@ import java.util.Iterator; import java.util.List; import org.springframework.batch.core.BatchStatus; -import org.springframework.batch.core.StartLimitExceededException; -import org.springframework.batch.core.UnexpectedJobExecutionException; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobExecutionException; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StartLimitExceededException; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.UnexpectedJobExecutionException; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.ExitStatus; @@ -87,10 +87,10 @@ public class SimpleJob extends AbstractJob { StepExecution lastStepExecution = getJobRepository().getLastStepExecution(jobInstance, step); - boolean isRestart = (getJobRepository().getStepExecutionCount(jobInstance, step) > 0 && !lastStepExecution - .getExitStatus().equals(ExitStatus.FINISHED)) ? true : false; + boolean isRestart = (lastStepExecution != null && !lastStepExecution.getStatus().equals( + BatchStatus.COMPLETED)) ? true : false; - if (isRestart && lastStepExecution != null) { + if (isRestart) { currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); } else { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java index e8c5ca4f9..7baf1d3db 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java @@ -346,7 +346,6 @@ public class SimpleJobTests extends TestCase { assertSame(exception, e); } assertTrue(step1.passedInStepContext.isEmpty()); - System.err.println(step2.passedInStepContext); assertFalse(step2.passedInStepContext.isEmpty()); } @@ -504,20 +503,24 @@ public class SimpleJobTests extends TestCase { if (exception instanceof RuntimeException) { stepExecution.setExitStatus(ExitStatus.FAILED); + stepExecution.setStatus(BatchStatus.FAILED); throw (RuntimeException) exception; } if (exception instanceof Error) { stepExecution.setExitStatus(ExitStatus.FAILED); + stepExecution.setStatus(BatchStatus.FAILED); throw (Error) exception; } if (exception instanceof JobInterruptedException) { stepExecution.setExitStatus(ExitStatus.FAILED); + stepExecution.setStatus(BatchStatus.FAILED); throw (JobInterruptedException) exception; } if (runnable != null) { runnable.run(); } stepExecution.setExitStatus(ExitStatus.FINISHED); + stepExecution.setStatus(BatchStatus.COMPLETED); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java index 1f8e71494..90246b0cd 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/support/RetryTemplate.java @@ -194,9 +194,9 @@ public class RetryTemplate implements RetryOperations { } /* - * An external policy that can retry should have rethrown the - * exception by now - i.e. we shouldn't get this far for an - * external policy if it can retry. + * A stateful policy that can retry should have rethrown the + * exception by now - i.e. we shouldn't get this far for a + * stateful policy if it can retry. */ } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java index 7be2807d1..c998f3a89 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java +++ b/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 message = new GenericMessage(request ); + GenericMessage message = new GenericMessage(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? diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java index 8c40446b3..11448f5d3 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java +++ b/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."); } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java index 9c194aba1..e7c8a3827 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java @@ -183,7 +183,7 @@ public class StepExecutionMessageHandlerTests { assertNotNull(message); assertEquals(1, jobExecution.getStepExecutions().size()); StepExecution stepExecution = (StepExecution) jobExecution.getStepExecutions().iterator().next(); - assertEquals(BatchStatus.STARTING, stepExecution.getStatus()); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); // We expect to get the context from the previous execution, even if we // do not execute assertTrue(stepExecution.getExecutionContext().containsKey("foo"));