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 25a77fedd..91838b0a9 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 @@ -48,6 +48,23 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo private long throttleLimit = DEFAULT_THROTTLE_LIMIT; + private int DEFAULT_MAX_WAIT_TIMEOUTS = 40; + + private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS; + + /** + * The maximum number of times to wait at the end of a step for a non-null + * result from the remote workers. This is a multiplier on the receive + * timeout set separately on the gateway. The ideal value is a compromise + * between allowing slow workers time to finish, and responsiveness if there + * is a dead worker. Defaults to 40. + * + * @param maxWaitTimeouts the maximum number of wait timeouts + */ + public void setMaxWaitTimeouts(int maxWaitTimeouts) { + this.maxWaitTimeouts = maxWaitTimeouts; + } + /** * Public setter for the throttle limit. This limits the number of pending * requests for chunk processing to avoid overwhelming the receivers. @@ -135,9 +152,8 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo * @return true if successfully received a result, false if timed out */ private boolean waitForResults() { - // TODO: cumulative timeout, or throw an exception? int count = 0; - int maxCount = 40; + int maxCount = maxWaitTimeouts; while (localState.getExpecting() > 0 && count++ < maxCount) { getNextResult(); } @@ -155,7 +171,6 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo * instance id (maybe we are sharing a channel and we shouldn't be) */ private void getNextResult() { - // TODO: make sure this is transactional (should be if single threaded) ChunkResponse payload = (ChunkResponse) messagingGateway.receive(); if (payload != null) { Long jobInstanceId = payload.getJobId(); diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java index 78253d7b0..d13153fd7 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java @@ -110,8 +110,7 @@ public class MessageOrientedStep extends AbstractStep { } else { executionContext.putString(WAITING, "true"); - // TODO: need these two lines to be atomic - getJobRepository().update(stepExecution); + getJobRepository().updateExecutionContext(stepExecution); outputChannel.send(new GenericMessage(request)); waitForReply(request.getJobId()); } @@ -130,7 +129,6 @@ public class MessageOrientedStep extends AbstractStep { while (count++ < maxCount) { - // TODO: timeout? @SuppressWarnings("unchecked") Message message = (Message) source.receive(timeout); @@ -146,8 +144,6 @@ public class MessageOrientedStep extends AbstractStep { // One of the steps decided we were finished // TODO: wait for all the other steps that might be // executing concurrently? - // TODO: maybe *any* reply on this channel should - // mean the end of the step? break; } 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 cccdc5fa1..1953f409f 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 @@ -17,13 +17,10 @@ package org.springframework.batch.integration.job; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobExecution; -import org.springframework.batch.core.JobExecutionException; -import org.springframework.batch.core.JobInstance; -import org.springframework.batch.core.StartLimitExceededException; import org.springframework.batch.core.Step; -import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.job.SimpleStepHandler; +import org.springframework.batch.core.job.StepHandler; import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.item.ExecutionContext; import org.springframework.beans.factory.annotation.Required; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; @@ -37,7 +34,7 @@ public class StepExecutionMessageHandler { private Step step; - private JobRepository jobRepository; + private StepHandler stepHandler; /** * Public setter for the {@link Step}. @@ -57,7 +54,7 @@ public class StepExecutionMessageHandler { */ @Required public void setJobRepository(JobRepository jobRepository) { - this.jobRepository = jobRepository; + stepHandler = new SimpleStepHandler(jobRepository); } @ServiceActivator @@ -69,43 +66,10 @@ public class StepExecutionMessageHandler { } JobExecution jobExecution = request.getJobExecution(); - JobInstance jobInstance = jobExecution.getJobInstance(); - StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); try { - - StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName()); - - // Even if it completed successfully we want to pass on the output - // attributes, so set up the execution context here if it is - // available. - if (lastStepExecution != null) { - stepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); - } - - // If it is already complete and not restartable it will simply be - // skipped - if (shouldStart(lastStepExecution, step)) { - - if (!isRestart(jobInstance, lastStepExecution)) { - stepExecution.setExecutionContext(new ExecutionContext()); - } - jobRepository.add(stepExecution); - 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()); - - } - + + stepHandler.handleStep(step, jobExecution); // (the job might actually not be complete, but the stage is). request.setStatus(BatchStatus.COMPLETED); @@ -122,15 +86,6 @@ 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 @@ -149,45 +104,4 @@ public class StepExecutionMessageHandler { request.setStatus(BatchStatus.FAILED); } - /* - * TODO: merge this with SimpleJob implementation. - * - * Given a step and configuration, return true if the step should start, - * false if it should not, and throw an exception if the job should finish. - */ - private boolean shouldStart(StepExecution lastStepExecution, Step step) throws JobExecutionException { - - BatchStatus stepStatus; - // if the last execution is null, the step has never been executed. - if (lastStepExecution == null) { - return true; - } - else { - stepStatus = lastStepExecution.getStatus(); - } - - if (stepStatus == BatchStatus.UNKNOWN) { - throw new JobExecutionException("Cannot restart step from UNKNOWN status. " - + "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."); - } - - if (stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) { - // step is complete, false should be returned, indicating that the - // step should not be started - return false; - } - - if (jobRepository.getStepExecutionCount(lastStepExecution.getJobExecution().getJobInstance(), step.getName()) < step - .getStartLimit()) { - // step start count is less than start max, return true - return true; - } - else { - // start max has been exceeded, throw an exception. - throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName() - + "StartMax: " + step.getStartLimit()); - } - } - } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/StepSupport.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/StepSupport.java index 7b02b7a13..f9a2cc114 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/StepSupport.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/StepSupport.java @@ -26,7 +26,7 @@ import org.springframework.batch.core.StepExecution; public class StepSupport implements Step { private String name; - private int startLimit; + private int startLimit = 1; /** * @param name diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java index 207223fd0..63c253dca 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java @@ -37,11 +37,11 @@ public class AsyncItemProcessorMessagingGatewayTests { for (Future future : list) { String value = future.get(); /** - * TODO: this delegate is a Spring Integration MessagingGateway. It - * can easily return null because of a timeout, but that will be - * treated by Batch as a filtered item, whereas it is really more - * like a skip. Maybe we should have an option to throw an exception - * in the processor if an unexpected null value comes back? + * This delegate is a Spring Integration MessagingGateway. It can + * easily return null because of a timeout, but that will be treated + * by Batch as a filtered item, whereas it is really more like a + * skip. So we have to throw an exception in the processor if an + * unexpected null value comes back. */ assertNotNull(value); assertTrue(value.matches("foo.*foo.*")); 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 c097fd5b1..d50e8d669 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 @@ -165,11 +165,6 @@ public class StepExecutionMessageHandlerTests { public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) { StepExecution stepExecution = new StepExecution(stepName, new JobExecution(jobInstance)); stepExecution.setStatus(BatchStatus.COMPLETED); - stepExecution.setExecutionContext(new ExecutionContext() { - { - put("foo", "bar"); - } - }); return stepExecution; } }; @@ -177,12 +172,8 @@ public class StepExecutionMessageHandlerTests { JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters()); JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); assertNotNull(message); - assertEquals(1, jobExecution.getStepExecutions().size()); - StepExecution stepExecution = (StepExecution) jobExecution.getStepExecutions().iterator().next(); - 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")); + assertEquals(0, jobExecution.getStepExecutions().size()); + assertEquals(BatchStatus.STARTING, jobExecution.getStatus()); } @Test @@ -203,7 +194,7 @@ public class StepExecutionMessageHandlerTests { JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters()); JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution)); assertNotNull(message); - assertEquals(1, jobExecution.getStepExecutions().size()); + assertEquals(0, jobExecution.getStepExecutions().size()); JobExecutionRequest payload = message; assertEquals(BatchStatus.FAILED, payload.getStatus()); assertTrue(payload.hasErrors()); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/BeanFactoryStepLocatorTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/BeanFactoryStepLocatorTests.java index c50aa27ba..049a2c897 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/BeanFactoryStepLocatorTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/BeanFactoryStepLocatorTests.java @@ -46,12 +46,10 @@ public class BeanFactoryStepLocatorTests { } public int getStartLimit() { - // TODO Auto-generated method stub return 0; } public boolean isAllowStartIfComplete() { - // TODO Auto-generated method stub return false; } }