class LocalState {

		private final AtomicInteger current = new AtomicInteger(-1);

		private final AtomicInteger actual = new AtomicInteger();

		private final AtomicInteger expected = new AtomicInteger();

		private final AtomicInteger redelivered = new AtomicInteger();

		private StepExecution stepExecution;

		private final Queue<ChunkResponse> contributions = new LinkedBlockingQueue<>();

		public int getExpecting() {
			return expected.get() - actual.get();
		}

		public <T> ChunkRequest<T> getRequest(Chunk<? 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<ChunkResponse> pollChunkResponses() {<FILL_FUNCTION_BODY>}

		public void pushResponse(ChunkResponse stepContribution) {
			synchronized (contributions) {
				contributions.add(stepContribution);
			}
		}

		public void incrementRedelivered() {
			redelivered.incrementAndGet();
		}

		public void incrementActual() {
			actual.incrementAndGet();
		}

		public void incrementExpected() {
			expected.incrementAndGet();
		}

		public StepContribution createStepContribution() {
			return stepExecution.createStepContribution();
		}

		public Long getJobId() {
			return stepExecution.getJobExecution().getJobId();
		}

		public void setStepExecution(StepExecution stepExecution) {
			this.stepExecution = stepExecution;
		}

		public void reset() {
			expected.set(0);
			actual.set(0);
		}

	}

			Collection<ChunkResponse> set = new ArrayList<>();
			synchronized (contributions) {
				ChunkResponse item = contributions.poll();
				while (item != null) {
					set.add(item);
					item = contributions.poll();
				}
			}
			return set;
		
<no_super_class>