@Override
    public void addOutputBuffers(List<OutputBufferId> newBuffers, boolean noMoreBuffers)
    {
        // All buffers are created in the constructor, so just validate that this isn't
        // a request to add a new buffer
        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);
            }
        }
    }
@Test
    public void test()
    {
        AtomicReference<OutputBuffers> outputBufferTarget = new AtomicReference<>();

        PartitionedOutputBufferManager hashOutputBufferManager = new PartitionedOutputBufferManager(FIXED_HASH_DISTRIBUTION, 4, outputBufferTarget::set);

        // output buffers are set immediately when the manager is created
        assertOutputBuffers(outputBufferTarget.get());

        // add buffers, which does not cause an error
        hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(0)), false);
        assertOutputBuffers(outputBufferTarget.get());
        hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(3)), true);
        assertOutputBuffers(outputBufferTarget.get());

        // try to a buffer out side of the partition range, which should result in an error
        try {
            hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(5)), false);
            fail("Expected IllegalStateException");
        }
        catch (IllegalStateException e) {
        }
        assertOutputBuffers(outputBufferTarget.get());

        // try to a buffer out side of the partition range, which should result in an error
        try {
            hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(6)), true);
            fail("Expected IllegalStateException");
        }
        catch (IllegalStateException e) {
        }
        assertOutputBuffers(outputBufferTarget.get());
    }