diff --git a/spring-batch-admin-parent/pom.xml b/spring-batch-admin-parent/pom.xml
index 5cf99ced9..3827b8726 100644
--- a/spring-batch-admin-parent/pom.xml
+++ b/spring-batch-admin-parent/pom.xml
@@ -346,6 +346,12 @@
1.2.14runtime
+
+ com.h2database
+ h2
+ 1.2.125
+ test
+ hsqldbhsqldb
diff --git a/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs b/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs
index 96349f772..edfc2175b 100644
--- a/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs
+++ b/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs
@@ -1,7 +1,7 @@
-#Fri Sep 03 15:05:09 BST 2010
+#Fri Sep 17 16:25:48 BST 2010
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml=\n\n\n\n\n\n
-//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJdbcIntegrationTests-context.xml=\n\n\n\n\n\n
+//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJdbcIntegrationTests-context.xml=\n\n\n\n\n\n
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml=\r\n\r\n\r\n\r\n\r\n\r\n
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/FileToMessagesJobIntegrationTests-context.xml=\r\n\r\n\r\n\r\n\r\n\r\n
@@ -9,5 +9,7 @@
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/step/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/tasklet/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/integration\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n\n\n\n\n
+//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/integration\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJdbcIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n\n\n\n\n
//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/integration\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml=\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n
+//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/integration\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n
eclipse.preferences.version=1
diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml
index 37ee82005..cfacfee60 100644
--- a/spring-batch-integration/pom.xml
+++ b/spring-batch-integration/pom.xml
@@ -1,5 +1,6 @@
-
+4.0.0spring-batch-integrationSpring Batch Integration
@@ -68,6 +69,21 @@
commons-dbcptest
+
+ com.h2database
+ h2
+ test
+
+
+ mysql
+ mysql-connector-java
+ test
+
+
+ org.apache.derby
+ derby
+ test
+ hsqldbhsqldb
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 a0608dc17..588a0fe63 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
@@ -64,11 +64,9 @@ public class ChunkMessageChannelItemWriter 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 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 extends StepExecutionListenerSuppo
if (!items.isEmpty()) {
- logger.debug("Dispatching chunk: " + items);
- ChunkRequest request = new ChunkRequest(items, localState.getJobId(), localState
- .createStepContribution());
+ ChunkRequest request = localState.getRequest(items);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dispatching chunk: " + request);
+ }
messagingGateway.send(new GenericMessage>(request));
localState.incrementExpected();
@@ -142,9 +141,6 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo
}
for (StepContribution contribution : getStepContributions()) {
- if (logger.isDebugEnabled()) {
- logger.debug("Applying:" + contribution);
- }
stepExecution.apply(contribution);
}
}
@@ -175,12 +171,19 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo
}
public Collection getStepContributions() {
- return localState.pollStepContributions();
+ List contributions = new ArrayList();
+ 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 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 message = messagingGateway.receive(replyChannel);
@@ -235,7 +236,7 @@ public class ChunkMessageChannelItemWriter 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 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 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 extends StepExecutionListenerSuppo
private StepExecution stepExecution;
- private Queue contributions = new LinkedBlockingQueue();
+ private Queue contributions = new LinkedBlockingQueue();
public int getExpecting() {
return expected.get() - actual.get();
}
+ public ChunkRequest getRequest(List extends T> items) {
+ return new ChunkRequest(current.incrementAndGet(), items, getJobId(), createStepContribution());
+ }
+
public void open(int expectedValue, int actualValue) {
actual.set(actualValue);
expected.set(expectedValue);
}
- public Collection pollStepContributions() {
- Collection set = new ArrayList();
+ public Collection pollChunkResponses() {
+ Collection set = new ArrayList();
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 extends StepExecutionListenerSuppo
return set;
}
- public void pushStepContribution(StepContribution stepContribution) {
+ public void pushResponse(ChunkResponse stepContribution) {
synchronized (contributions) {
contributions.add(stepContribution);
}
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
index cb6dda7b5..9df1b75dd 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
@@ -80,12 +80,12 @@ public class ChunkProcessorChunkHandler implements ChunkHandler, 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);
}
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java
index a295614b1..16655f2cd 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java
@@ -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 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 implements Serializable {
return items;
}
+ public int getSequence() {
+ return sequence;
+ }
+
/**
* @return the {@link StepContribution} for this chunk
*/
@@ -62,8 +70,8 @@ public class ChunkRequest 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();
}
}
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java
index 46a21dbf5..a422e7705 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java
@@ -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;
}
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java
index 71dfc3197..a6781012a 100644
--- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java
@@ -195,7 +195,7 @@ public class ChunkMessageItemWriterIntegrationTests {
private GenericMessage getSimpleMessage(String string, Long jobId) {
StepContribution stepContribution = new JobExecution(new JobInstance(0L, new JobParameters(), "job"), 1L)
.createStepExecution("step").createStepContribution();
- ChunkRequest chunk = new ChunkRequest(StringUtils.commaDelimitedListToSet(string), jobId, stepContribution);
+ ChunkRequest chunk = new ChunkRequest(0, StringUtils.commaDelimitedListToSet(string), jobId, stepContribution);
GenericMessage message = new GenericMessage(chunk);
return message;
}
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java
index 908d51ea2..549e249a2 100644
--- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java
@@ -24,8 +24,8 @@ public class ChunkProcessorChunkHandlerTests {
}
});
StepContribution stepContribution = MetaDataInstanceFactory.createStepExecution().createStepContribution();
- ChunkResponse response = handler.handleChunk(new ChunkRequest