class ChunkMonitorData {

		public int offset;

		public int chunkSize;

		public ChunkMonitorData(int offset, int chunkSize) {
			this.offset = offset;
			this.chunkSize = chunkSize;
		}

	}

	private static final String OFFSET = "OFFSET";

	private final CompositeItemStream stream = new CompositeItemStream();

	private final ThreadLocal<ChunkMonitorData> holder = new ThreadLocal<>();

	private ItemReader<?> reader;

	public ChunkMonitor() {
		this.setExecutionContextName(ChunkMonitor.class.getName());
	}

	/**
	 * @param stream the stream to set
	 */
	public void registerItemStream(ItemStream stream) {
		streamsRegistered = true;
		this.stream.register(stream);
	}

	/**
	 * @param reader the reader to set
	 */
	public void setItemReader(ItemReader<?> reader) {
		this.reader = reader;
	}

	public void incrementOffset() {
		ChunkMonitorData data = getData();
		data.offset++;
		if (data.offset >= data.chunkSize) {
			resetOffset();
		}
	}

	public int getOffset() {
		return getData().offset;
	}

	public void resetOffset() {
		getData().offset = 0;
	}

	public void setChunkSize(int chunkSize) {
		getData().chunkSize = chunkSize;
		resetOffset();
	}

	@Override
	public void close() throws ItemStreamException {
		super.close();
		holder.set(null);
		if (streamsRegistered) {
			stream.close();
		}
	}

	@Override
	public void open(ExecutionContext executionContext) throws ItemStreamException {
		super.open(executionContext);
		if (streamsRegistered) {
			stream.open(executionContext);
			ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(getExecutionContextKey(OFFSET), 0), 0);
			holder.set(data);
			if (reader == null) {
				logger.warn("No ItemReader set (must be concurrent step), so ignoring offset data.");
				return;
			}
			for (int i = 0; i < data.offset; i++) {
				try {
					reader.read();
				}
				catch (Exception e) {
					throw new ItemStreamException("Could not position reader with offset: " + data.offset, e);
				}
			}

			resetOffset();
		}
	}

	@Override
	public void update(ExecutionContext executionContext) throws ItemStreamException {
		super.update(executionContext);
		if (streamsRegistered) {
			ChunkMonitorData data = getData();
			if (data.offset == 0) {
				// Only call the underlying update method if we are on a chunk
				// boundary
				stream.update(executionContext);
				executionContext.remove(getExecutionContextKey(OFFSET));
			}
			else {
				executionContext.putInt(getExecutionContextKey(OFFSET), data.offset);
			}
		}
	}

	private ChunkMonitorData getData() {<FILL_FUNCTION_BODY>

		ChunkMonitorData data = holder.get();
		if (data == null) {
			if (streamsRegistered) {
				logger.warn("ItemStream was opened in a different thread.  Restart data could be compromised.");
			}
			data = new ChunkMonitorData(0, 0);
			holder.set(data);
		}
		return data;
	
<methods>public non-sealed void <init>() ,public void close() ,public java.lang.String getExecutionContextKey(java.lang.String) ,public java.lang.String getName() ,public void open(org.springframework.batch.item.ExecutionContext) ,public void setName(java.lang.String) ,public void update(org.springframework.batch.item.ExecutionContext) <variables>private final org.springframework.batch.item.util.ExecutionContextUserSupport executionContextUserSupport