class BroadcastOutputBufferManager implements OutputBufferManager {

    @Override
    public void addOutputBuffers(List<OutputBufferId> newBuffers, boolean noMoreBuffers)
    {
        OutputBuffers newOutputBuffers;
        synchronized (this) {
            if (outputBuffers.isNoMoreBufferIds()) {
                // a stage can move to a final state (e.g., failed) while scheduling, so ignore
                // the new buffers
                return;
            }

            OutputBuffers originalOutputBuffers = outputBuffers;

            // Note: it does not matter which partition id the task is using, in broadcast all tasks read from the same partition
            for (OutputBufferId newBuffer : newBuffers) {
                outputBuffers = outputBuffers.withBuffer(newBuffer, BROADCAST_PARTITION_ID);
            }

            if (noMoreBuffers) {
                outputBuffers = outputBuffers.withNoMoreBufferIds();
            }

            // don't update if nothing changed
            if (outputBuffers == originalOutputBuffers) {
                return;
            }
            newOutputBuffers = this.outputBuffers;
        }
        outputBufferTarget.accept(newOutputBuffers);
    }

    public  BroadcastOutputBufferManager(Consumer<OutputBuffers> outputBufferTarget);

}