### Question:
BroadcastOutputBuffer implements OutputBuffer { @VisibleForTesting OutputBufferMemoryManager getMemoryManager() { return memoryManager; } 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 testSharedBufferBlocking() { SettableFuture<?> blockedFuture = SettableFuture.create(); MockMemoryReservationHandler reservationHandler = new MockMemoryReservationHandler(blockedFuture); AggregatedMemoryContext memoryContext = newRootAggregatedMemoryContext(reservationHandler, 0L); Page page = createPage(1); long pageSize = PAGES_SERDE.serialize(page).getRetainedSizeInBytes(); BroadcastOutputBuffer buffer = createBroadcastBuffer(createInitialEmptyOutputBuffers(BROADCAST), DataSize.ofBytes(pageSize * 2), memoryContext, directExecutor()); OutputBufferMemoryManager memoryManager = buffer.getMemoryManager(); enqueuePage(buffer, page); blockedFuture.set(null); memoryManager.onMemoryAvailable(); assertTrue(memoryManager.getBufferBlockedFuture().isDone(), "buffer shouldn't be blocked"); addPage(buffer, page); enqueuePage(buffer, page); }
@Test public void testSharedBufferBlocking2() { SettableFuture<?> blockedFuture = SettableFuture.create(); blockedFuture.set(null); MockMemoryReservationHandler reservationHandler = new MockMemoryReservationHandler(blockedFuture); AggregatedMemoryContext memoryContext = newRootAggregatedMemoryContext(reservationHandler, 0L); Page page = createPage(1); long pageSize = PAGES_SERDE.serialize(page).getRetainedSizeInBytes(); BroadcastOutputBuffer buffer = createBroadcastBuffer(createInitialEmptyOutputBuffers(BROADCAST), DataSize.ofBytes(pageSize * 2), memoryContext, directExecutor()); OutputBufferMemoryManager memoryManager = buffer.getMemoryManager(); addPage(buffer, page); addPage(buffer, page); blockedFuture = SettableFuture.create(); reservationHandler.updateBlockedFuture(blockedFuture); memoryManager.updateMemoryUsage(1L); blockedFuture.set(null); memoryManager.onMemoryAvailable(); assertFalse(memoryManager.getBufferBlockedFuture().isDone(), "buffer should be blocked"); memoryManager.updateMemoryUsage(-pageSize * 2 - 1); assertTrue(memoryManager.getBufferBlockedFuture().isDone(), "buffer shouldn't be blocked"); addPage(buffer, page); addPage(buffer, page); enqueuePage(buffer, page); }
@Test public void testSharedBufferBlockingNoBlockOnFull() { SettableFuture<?> blockedFuture = SettableFuture.create(); MockMemoryReservationHandler reservationHandler = new MockMemoryReservationHandler(blockedFuture); AggregatedMemoryContext memoryContext = newRootAggregatedMemoryContext(reservationHandler, 0L); Page page = createPage(1); long pageSize = PAGES_SERDE.serialize(page).getRetainedSizeInBytes(); BroadcastOutputBuffer buffer = createBroadcastBuffer(createInitialEmptyOutputBuffers(BROADCAST), DataSize.ofBytes(pageSize * 2), memoryContext, directExecutor()); OutputBufferMemoryManager memoryManager = buffer.getMemoryManager(); memoryManager.setNoBlockOnFull(); enqueuePage(buffer, page); blockedFuture.set(null); memoryManager.onMemoryAvailable(); assertTrue(memoryManager.getBufferBlockedFuture().isDone(), "buffer shouldn't be blocked"); addPage(buffer, page); addPage(buffer, page); }