### Question:
BroadcastOutputBufferManager implements OutputBufferManager { @Override public void addOutputBuffers(List<OutputBufferId> newBuffers, boolean noMoreBuffers) { OutputBuffers newOutputBuffers; synchronized (this) { if (outputBuffers.isNoMoreBufferIds()) { return; } OutputBuffers originalOutputBuffers = outputBuffers; for (OutputBufferId newBuffer : newBuffers) { outputBuffers = outputBuffers.withBuffer(newBuffer, BROADCAST_PARTITION_ID); } if (noMoreBuffers) { outputBuffers = outputBuffers.withNoMoreBufferIds(); } if (outputBuffers == originalOutputBuffers) { return; } newOutputBuffers = this.outputBuffers; } outputBufferTarget.accept(newOutputBuffers); } BroadcastOutputBufferManager(Consumer<OutputBuffers> outputBufferTarget); @Override void addOutputBuffers(List<OutputBufferId> newBuffers, boolean noMoreBuffers);  }### Answer:
@Test public void test() { AtomicReference<OutputBuffers> outputBufferTarget = new AtomicReference<>(); BroadcastOutputBufferManager hashOutputBufferManager = new BroadcastOutputBufferManager(outputBufferTarget::set); assertEquals(outputBufferTarget.get(), createInitialEmptyOutputBuffers(BROADCAST)); hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(0)), false); OutputBuffers expectedOutputBuffers = createInitialEmptyOutputBuffers(BROADCAST).withBuffer(new OutputBufferId(0), BROADCAST_PARTITION_ID); assertEquals(outputBufferTarget.get(), expectedOutputBuffers); hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(1), new OutputBufferId(2)), false); expectedOutputBuffers = expectedOutputBuffers.withBuffer(new OutputBufferId(1), BROADCAST_PARTITION_ID); expectedOutputBuffers = expectedOutputBuffers.withBuffer(new OutputBufferId(2), BROADCAST_PARTITION_ID); assertEquals(outputBufferTarget.get(), expectedOutputBuffers); hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(3)), true); expectedOutputBuffers = expectedOutputBuffers.withBuffer(new OutputBufferId(3), BROADCAST_PARTITION_ID); expectedOutputBuffers = expectedOutputBuffers.withNoMoreBufferIds(); assertEquals(outputBufferTarget.get(), expectedOutputBuffers); hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(5)), false); assertEquals(outputBufferTarget.get(), expectedOutputBuffers); hashOutputBufferManager.addOutputBuffers(ImmutableList.of(new OutputBufferId(6)), true); assertEquals(outputBufferTarget.get(), expectedOutputBuffers); }