### Question:
BroadcastOutputBuffer implements OutputBuffer { @Override public ListenableFuture<BufferResult> get(OutputBufferId outputBufferId, long startingSequenceId, DataSize maxSize) { checkState(!Thread.holdsLock(this), "Cannot get pages while holding a lock on this"); requireNonNull(outputBufferId, "outputBufferId is null"); checkArgument(maxSize.toBytes() > 0, "maxSize must be at least 1 byte"); return getBuffer(outputBufferId).getPages(startingSequenceId, maxSize); } BroadcastOutputBuffer(
            String taskInstanceId,
            StateMachine<BufferState> state,
            DataSize maxBufferSize,
            Supplier<LocalMemoryContext> systemMemoryContextSupplier,
            Executor notificationExecutor,
            Runnable notifyStatusChanged); @Override void addStateChangeListener(StateChangeListener<BufferState> stateChangeListener); @Override boolean isFinished(); @Override double getUtilization(); @Override boolean isOverutilized(); @Override OutputBufferInfo getInfo(); @Override void setOutputBuffers(OutputBuffers newOutputBuffers); @Override ListenableFuture<?> isFull(); @Override void enqueue(List<SerializedPage> pages); @Override void enqueue(int partitionNumber, List<SerializedPage> pages); @Override ListenableFuture<BufferResult> get(OutputBufferId outputBufferId, long startingSequenceId, DataSize maxSize); @Override void acknowledge(OutputBufferId bufferId, long sequenceId); @Override void abort(OutputBufferId bufferId); @Override void setNoMorePages(); @Override void destroy(); @Override void fail(); @Override long getPeakMemoryUsage();  }### Answer:
@Test public void testDuplicateRequests() { BroadcastOutputBuffer buffer = createBroadcastBuffer( createInitialEmptyOutputBuffers(BROADCAST) .withBuffer(FIRST, BROADCAST_PARTITION_ID) .withNoMoreBufferIds(), sizeOfPages(10)); for (int i = 0; i < 3; i++) { addPage(buffer, createPage(i)); } assertQueueState(buffer, FIRST, 3, 0); assertBufferResultEquals(TYPES, getBufferResult(buffer, FIRST, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0), createPage(1), createPage(2))); assertQueueState(buffer, FIRST, 3, 0); assertBufferResultEquals(TYPES, getBufferResult(buffer, FIRST, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0), createPage(1), createPage(2))); assertQueueState(buffer, FIRST, 3, 0); buffer.get(FIRST, 3, sizeOfPages(10)).cancel(true); assertBufferResultEquals(TYPES, getBufferResult(buffer, FIRST, 0, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, false)); assertQueueState(buffer, FIRST, 0, 3); }
@Test public void testAcknowledgementFreesWriters() { BroadcastOutputBuffer buffer = createBroadcastBuffer( createInitialEmptyOutputBuffers(BROADCAST) .withBuffer(FIRST, BROADCAST_PARTITION_ID) .withBuffer(SECOND, BROADCAST_PARTITION_ID) .withNoMoreBufferIds(), sizeOfPages(2)); addPage(buffer, createPage(1)); addPage(buffer, createPage(2)); assertQueueState(buffer, FIRST, 2, 0); ListenableFuture<?> future = enqueuePage(buffer, createPage(3)); assertFalse(future.isDone()); assertQueueState(buffer, FIRST, 3, 0); assertQueueState(buffer, SECOND, 3, 0); buffer.get(FIRST, 2, sizeOfPages(10)).cancel(true); assertFalse(future.isDone()); buffer.get(SECOND, 2, sizeOfPages(10)).cancel(true); assertFutureIsDone(future); assertQueueState(buffer, SECOND, 1, 2); }