Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Tidy up some TODOs

Browse Source
This commit is contained in:
David Syer
2010-01-05 16:37:44 +00:00
committed by Michael Minella
parent 0da5f48895
commit c30a58bfb5
7 changed files with 34 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -48,6 +48,23 @@ public class ChunkMessageChannelItemWriter<T> 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<T> 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<T> 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();

6
spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java
View File

@@ -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<JobExecutionRequest>(request));
waitForReply(request.getJobId());
}
@@ -130,7 +129,6 @@ public class MessageOrientedStep extends AbstractStep {
while (count++ < maxCount) {
// TODO: timeout?
@SuppressWarnings("unchecked")
Message<JobExecutionRequest> message = (Message<JobExecutionRequest>) 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;
}

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

@@ -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());
}
}
}