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

BATCHADM-72: Improve chunking traceability

Browse Source 
- add sequence info to chunks
- Improve logging in ChunkMessageChannelItemWriter
- Add mysql/h2/derby properties to integration project
- Add drop scripts necessary for persistent databases
This commit is contained in:
Dave Syer
2010-09-17 16:57:31 +01:00
committed by Michael Minella
parent a00fd31172
commit 88638ce05f
19 changed files with 208 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -64,11 +64,9 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
private PollableChannel replyChannel;
/**
* 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.
* 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
*/
@@ -77,8 +75,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
/**
* Public setter for the throttle limit. This limits the number of pending
* requests for chunk processing to avoid overwhelming the receivers.
* 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) {
@@ -102,9 +100,10 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
if (!items.isEmpty()) {
logger.debug("Dispatching chunk: " + items);
ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState
.createStepContribution());
ChunkRequest<T> request = localState.getRequest(items);
if (logger.isDebugEnabled()) {
logger.debug("Dispatching chunk: " + request);
}
messagingGateway.send(new GenericMessage<ChunkRequest<T>>(request));
localState.incrementExpected();
@@ -142,9 +141,6 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
for (StepContribution contribution : getStepContributions()) {
if (logger.isDebugEnabled()) {
logger.debug("Applying:" + contribution);
}
stepExecution.apply(contribution);
}
}
@@ -175,12 +171,19 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
public Collection<StepContribution> getStepContributions() {
return localState.pollStepContributions();
List<StepContribution> contributions = new ArrayList<StepContribution>();
for (ChunkResponse response : localState.pollChunkResponses()) {
StepContribution contribution = response.getStepContribution();
if (logger.isDebugEnabled()) {
logger.debug("Applying: " + response);
}
contributions.add(contribution);
}
return contributions;
}
/**
* Wait until all the results that are in the pipeline come back to the
* reply channel.
* Wait until all the results that are in the pipeline come back to the reply channel.
*
* @return true if successfully received a result, false if timed out
*/
@@ -206,14 +209,12 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
/**
* Get the next result if it is available (within the timeout specified in
* the gateway), otherwise do nothing.
* Get the next result if it is available (within the timeout specified in the gateway), otherwise do nothing.
*
* @throws AsynchronousFailureException If there is a response and it
* contains a failed chunk response.
* @throws AsynchronousFailureException If there is a response and it contains a failed chunk response.
*
* @throws IllegalStateException if the result contains the wrong job
* instance id (maybe we are sharing a channel and we shouldn't be)
* @throws IllegalStateException if the result contains the wrong job instance id (maybe we are sharing a channel
* and we shouldn't be)
*/
private void getNextResult() throws AsynchronousFailureException {
Message<ChunkResponse> message = messagingGateway.receive(replyChannel);
@@ -235,7 +236,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
+ "is probably inconsistent, and the step will fail.");
localState.incrementRedelivered();
}
localState.pushStepContribution(payload.getStepContribution());
localState.pushResponse(payload);
localState.incrementActual();
if (!payload.isSuccessful()) {
throw new AsynchronousFailureException("Failure or interrupt detected in handler: "
@@ -245,8 +246,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
/**
* Re-throws the original throwable if it is unchecked, wraps checked
* exceptions into {@link AsynchronousFailureException}.
* Re-throws the original throwable if it is unchecked, wraps checked exceptions into
* {@link AsynchronousFailureException}.
*/
private static AsynchronousFailureException wrapIfNecessary(Throwable throwable) {
if (throwable instanceof Error) {
@@ -262,6 +263,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
private static class LocalState {
private AtomicInteger current = new AtomicInteger(-1);
private AtomicInteger actual = new AtomicInteger();
private AtomicInteger expected = new AtomicInteger();
@@ -270,21 +273,25 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
private StepExecution stepExecution;
private Queue<StepContribution> contributions = new LinkedBlockingQueue<StepContribution>();
private Queue<ChunkResponse> contributions = new LinkedBlockingQueue<ChunkResponse>();
public int getExpecting() {
return expected.get() - actual.get();
}
public <T> ChunkRequest<T> getRequest(List<? extends T> items) {
return new ChunkRequest<T>(current.incrementAndGet(), items, getJobId(), createStepContribution());
}
public void open(int expectedValue, int actualValue) {
actual.set(actualValue);
expected.set(expectedValue);
}
public Collection<StepContribution> pollStepContributions() {
Collection<StepContribution> set = new ArrayList<StepContribution>();
public Collection<ChunkResponse> pollChunkResponses() {
Collection<ChunkResponse> set = new ArrayList<ChunkResponse>();
synchronized (contributions) {
StepContribution item = contributions.poll();
ChunkResponse item = contributions.poll();
while (item != null) {
set.add(item);
item = contributions.poll();
@@ -293,7 +300,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
return set;
}
public void pushStepContribution(StepContribution stepContribution) {
public void pushResponse(ChunkResponse stepContribution) {
synchronized (contributions) {
contributions.add(stepContribution);
}

4
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
View File

@@ -80,12 +80,12 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
Throwable failure = process(chunkRequest, stepContribution);
if (failure != null) {
logger.debug("Failed chunk", failure);
return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, failure.getClass().getName()
return new ChunkResponse(false, chunkRequest.getSequence(), chunkRequest.getJobId(), stepContribution, failure.getClass().getName()
+ ": " + failure.getMessage());
}
logger.debug("Completed chunk handling with " + stepContribution);
return new ChunkResponse(true, chunkRequest.getJobId(), stepContribution);
return new ChunkResponse(true, chunkRequest.getSequence(), chunkRequest.getJobId(), stepContribution);
}

20
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java
View File

@@ -22,7 +22,8 @@ import java.util.Collection;
import org.springframework.batch.core.StepContribution;
/**
* Encapsulation of a chunk of items to be processed remotely as part of a step execution.
* Encapsulation of a chunk of items to be processed remotely as part of a step
* execution.
*
* @author Dave Syer
*
@@ -30,19 +31,22 @@ import org.springframework.batch.core.StepContribution;
*/
public class ChunkRequest<T> implements Serializable {
private final Long jobId;
private final long jobId;
private final Collection<? extends T> items;
private final StepContribution stepContribution;
public ChunkRequest(Collection<? extends T> items, Long jobId, StepContribution stepContribution) {
private final int sequence;
public ChunkRequest(int sequence, Collection<? extends T> items, long jobId, StepContribution stepContribution) {
this.sequence = sequence;
this.items = items;
this.jobId = jobId;
this.stepContribution = stepContribution;
}
public Long getJobId() {
public long getJobId() {
return jobId;
}
@@ -50,6 +54,10 @@ public class ChunkRequest<T> implements Serializable {
return items;
}
public int getSequence() {
return sequence;
}
/**
* @return the {@link StepContribution} for this chunk
*/
@@ -62,8 +70,8 @@ public class ChunkRequest<T> implements Serializable {
*/
@Override
public String toString() {
return getClass().getSimpleName() + ": jobId=" + jobId + ", contribution=" + stepContribution + ", item count="
+ items.size();
return getClass().getSimpleName() + ": jobId=" + jobId + ", sequence=" + sequence + ", contribution="
+ stepContribution + ", item count=" + items.size();
}
}

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

@@ -38,24 +38,27 @@ public class ChunkResponse implements Serializable {
private final boolean redelivered;
public ChunkResponse(Long jobId, StepContribution stepContribution) {
this(true, jobId, stepContribution, null);
private final int sequence;
public ChunkResponse(int sequence, Long jobId, StepContribution stepContribution) {
this(true, sequence, jobId, stepContribution, null);
}
public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution) {
this(status, jobId, stepContribution, null);
public ChunkResponse(boolean status, int sequence, Long jobId, StepContribution stepContribution) {
this(status, sequence, jobId, stepContribution, null);
}
public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message) {
this(status, jobId, stepContribution, message, false);
public ChunkResponse(boolean status, int sequence, Long jobId, StepContribution stepContribution, String message) {
this(status, sequence, jobId, stepContribution, message, false);
}
public ChunkResponse(ChunkResponse input, boolean redelivered) {
this(input.status, input.jobId, input.stepContribution, input.message, redelivered);
this(input.status, input.sequence, input.jobId, input.stepContribution, input.message, redelivered);
}
public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message, boolean redelivered) {
public ChunkResponse(boolean status, int sequence, Long jobId, StepContribution stepContribution, String message, boolean redelivered) {
this.status = status;
this.sequence = sequence;
this.jobId = jobId;
this.stepContribution = stepContribution;
this.message = message;
@@ -69,6 +72,10 @@ public class ChunkResponse implements Serializable {
public Long getJobId() {
return jobId;
}
public int getSequence() {
return sequence;
}
public boolean isSuccessful() {
return status;
@@ -87,7 +94,7 @@ public class ChunkResponse implements Serializable {
*/
@Override
public String toString() {
return getClass().getSimpleName() + ": jobId=" + jobId + ", stepContribution=" + stepContribution
return getClass().getSimpleName() + ": jobId=" + jobId + ", sequence=" + sequence + ", stepContribution=" + stepContribution
+ ", successful=" + status;
}