@VisibleForTesting
    OutputBufferMemoryManager getMemoryManager()
    {
        return memoryManager;
    }
@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();

        // create a buffer that can only hold two pages
        BroadcastOutputBuffer buffer = createBroadcastBuffer(createInitialEmptyOutputBuffers(BROADCAST), new DataSize(pageSize * 2, BYTE), memoryContext, directExecutor());
        OutputBufferMemoryManager memoryManager = buffer.getMemoryManager();

        memoryManager.setNoBlockOnFull();

        // even if setNoBlockOnFull() is called the buffer should block on memory when we add the first page
        // as no memory is available (MockMemoryReservationHandler will return a future that is not done)
        enqueuePage(buffer, page);

        // more memory is available
        blockedFuture.set(null);
        memoryManager.onMemoryAvailable();
        assertTrue(memoryManager.getBufferBlockedFuture().isDone(), "buffer shouldn't be blocked");

        // we should be able to add one more page after more memory is available
        addPage(buffer, page);

        // the buffer is full now, but setNoBlockOnFull() is called so the buffer shouldn't block
        addPage(buffer, page);
    }