@VisibleForTesting
    OutputBufferMemoryManager getMemoryManager()
    {
        return memoryManager;
    }
@Test
    public void testSharedBufferBlocking2()
    {
        // start with a complete future
        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();

        // 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();

        // add two pages to fill up the buffer (memory is available)
        addPage(buffer, page);
        addPage(buffer, page);

        // fill up the memory pool
        blockedFuture = SettableFuture.create();
        reservationHandler.updateBlockedFuture(blockedFuture);

        // allocate one more byte to make the buffer full
        memoryManager.updateMemoryUsage(1L);

        // more memory is available
        blockedFuture.set(null);
        memoryManager.onMemoryAvailable();

        // memoryManager should still return a blocked future as the buffer is still full
        assertFalse(memoryManager.getBufferBlockedFuture().isDone(), "buffer should be blocked");

        // remove all pages from the memory manager and the 1 byte that we added above
        memoryManager.updateMemoryUsage(-pageSize * 2 - 1);

        // now we have both buffer space and memory available, so memoryManager shouldn't be blocked
        assertTrue(memoryManager.getBufferBlockedFuture().isDone(), "buffer shouldn't be blocked");

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

        // the buffer is full now
        enqueuePage(buffer, page);
    }