BATCH-1444: better check for ChunkMonitor stream

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java

@@ -35,6 +35,8 @@ import org.springframework.batch.item.support.CompositeItemStream;
 class ChunkMonitor implements ItemStream {
 	
 	private Log logger = LogFactory.getLog(getClass());
+	
+	private boolean streamsRegistered = false;
 
 	public static class ChunkMonitorData {
 		public int offset;
@@ -64,6 +66,7 @@ class ChunkMonitor implements ItemStream {
 	 * @param stream the stream to set
 	 */
 	public void registerItemStream(ItemStream stream) {
+		streamsRegistered = true;
 		this.stream.register(stream);
 	}
 
@@ -97,13 +100,13 @@ class ChunkMonitor implements ItemStream {
 
 	public void close() throws ItemStreamException {
 		holder.set(new ChunkMonitorData(0,0));
-		if (stream != null) {
+		if (streamsRegistered) {
 			stream.close();
 		}
 	}
 
 	public void open(ExecutionContext executionContext) throws ItemStreamException {
-		if (stream != null) {
+		if (streamsRegistered) {
 			stream.open(executionContext);
 			ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(OFFSET, 0), 0);
 			holder.set(data);
@@ -123,7 +126,7 @@ class ChunkMonitor implements ItemStream {
 	}
 
 	public void update(ExecutionContext executionContext) throws ItemStreamException {
-		if (stream != null) {
+		if (streamsRegistered) {
 			ChunkMonitorData data = getData();
 			if (data.offset == 0) {
 				// Only call the underlying update method if we are on a chunk
@@ -139,7 +142,7 @@ class ChunkMonitor implements ItemStream {
 	private ChunkMonitorData getData() {
 		ChunkMonitorData data = holder.get();
 		if (data==null) {
-			if (stream!=null) {
+			if (streamsRegistered) {
 				logger.warn("ItemStream was opened in a different thread.  Restart data could be compromised.");
 			}
 			data = new ChunkMonitorData(0,0);

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java

@@ -134,4 +134,18 @@ public class ChunkMonitorTests {
 		assertEquals(1, executionContext.size());
 	}
 
+	@Test
+	public void testUpdateWithNoStream() throws Exception {
+		monitor = new ChunkMonitor();
+		monitor.setItemReader(new ItemReader<String>() {
+			public String read() throws Exception, UnexpectedInputException, ParseException {
+				return "" + (count++);
+			}
+		});
+		monitor.setChunkSize(CHUNK_SIZE);
+		monitor.incrementOffset();
+		ExecutionContext executionContext = new ExecutionContext();
+		monitor.update(executionContext);
+		assertEquals(0, executionContext.size());		
+	}
 }