### Question:
PartitionedOutputBufferManager implements OutputBufferManager { @Override public void addOutputBuffers(List<OutputBufferId> newBuffers, boolean noMoreBuffers) { for (OutputBufferId newBuffer : newBuffers) { Integer existingBufferId = outputBuffers.get(newBuffer); if (existingBufferId == null) { throw new IllegalStateException("Unexpected new output buffer " + newBuffer); } if (newBuffer.getId() != existingBufferId) { throw new IllegalStateException("newOutputBuffers has changed the assignment for task " + newBuffer); } } } PartitionedOutputBufferManager(PartitioningHandle partitioningHandle, int partitionCount, Consumer<OutputBuffers> outputBufferTarget); @Override void addOutputBuffers(List<OutputBufferId> newBuffers, boolean noMoreBuffers);  }### Answer:
@Test public void test() { AtomicReference<OutputBuffers> outputBufferTarget = new AtomicReference<>(); PartitionedOutputBufferManager hashOutputBufferManager = new PartitionedOutputBufferManager(FIXED_HASH_DISTRIBUTION, 4, outputBufferTarget::set); assertOutputBuffers(outputBufferTarget.get()); hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(0)), false); assertOutputBuffers(outputBufferTarget.get()); hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(3)), true); assertOutputBuffers(outputBufferTarget.get()); try { hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(5)), false); fail("Expected IllegalStateException"); } catch (IllegalStateException e) { } assertOutputBuffers(outputBufferTarget.get()); try { hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(6)), true); fail("Expected IllegalStateException"); } catch (IllegalStateException e) { } assertOutputBuffers(outputBufferTarget.get()); }