diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml
index 6001f2af3..e3f0c8af3 100644
--- a/spring-batch-integration/pom.xml
+++ b/spring-batch-integration/pom.xml
@@ -107,6 +107,12 @@
${spring.integration.version}
compile
+
+ org.springframework.integration
+ spring-integration-jms
+ ${spring.integration.version}
+ compile
+
org.springframework
spring-context
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 fafdd6736..25a77fedd 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
@@ -145,10 +145,17 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo
}
/**
- * Get the next result if it is available within the timeout specified,
- * otherwise return null.
+ * 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 IllegalStateException if the result contains the wrong job
+ * 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/test/java/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests.java
new file mode 100644
index 000000000..b538e91d6
--- /dev/null
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests.java
@@ -0,0 +1,36 @@
+package org.springframework.batch.integration.chunk;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class ChunkStepIntegrationTests {
+
+ @Autowired
+ private JobLauncher jobLauncher;
+
+ @Autowired
+ private Job job;
+
+ @Test
+ public void testOpenWithNoState() throws Exception {
+ JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
+ assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
+ StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
+ assertEquals(9, stepExecution.getReadCount());
+ assertEquals(9, stepExecution.getWriteCount());
+ }
+
+}
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemReader.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemReader.java
new file mode 100644
index 000000000..50883237a
--- /dev/null
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemReader.java
@@ -0,0 +1,70 @@
+package org.springframework.batch.integration.chunk;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ParseException;
+import org.springframework.batch.item.UnexpectedInputException;
+import org.springframework.stereotype.Component;
+
+@Component
+public class TestItemReader implements ItemReader {
+
+ private static final Log logger = LogFactory.getLog(TestItemReader.class);
+
+ /**
+ * Counts the number of chunks processed in the handler.
+ */
+ public volatile static int count = 0;
+
+ /**
+ * Item that causes failure in handler.
+ */
+ public final static String FAIL_ON = "bad";
+
+ /**
+ * Item that causes handler to wait to simulate delayed processing.
+ */
+ public static final String WAIT_ON = "wait";
+
+ private List items = new ArrayList();
+
+ /**
+ * @param items the items to set
+ */
+ public void setItems(List items) {
+ this.items = items;
+ }
+
+ public T read() throws Exception, UnexpectedInputException, ParseException {
+
+ if (count>=items.size()) {
+ return null;
+ }
+
+ T item = items.get(count++);
+
+ logger.debug("Reading "+item);
+
+ if (item.equals(WAIT_ON)) {
+ try {
+ Thread.sleep(200);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Unexpected interruption.", e);
+ }
+ }
+
+ if (item.equals(FAIL_ON)) {
+ throw new IllegalStateException("Planned failure on: " + FAIL_ON);
+ }
+
+ return item;
+
+ }
+
+}
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml
new file mode 100644
index 000000000..781477457
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml
@@ -0,0 +1,75 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ 1
+ 2
+ 3
+ 4
+ 5
+ 6
+ 7
+ 8
+ 9
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file