@Override
    public synchronized ListenableFuture<BufferResult> get(OutputBufferId bufferId, long startSequenceId, DataSize maxSize)
    {
        requireNonNull(bufferId, "outputBufferId is null");
        checkArgument(bufferId.getId() == outputBufferId.getId(), "Invalid buffer id");
        checkArgument(maxSize.toBytes() > 0, "maxSize must be at least 1 byte");

        acknowledge(bufferId, startSequenceId);

        long currentSequenceId = this.currentSequenceId.get();

        // process the request if we have no more data coming in, have data to read, or if this is an outdated request
        if (noMorePages.get() || !handleInfoQueue.isEmpty() || !pages.isEmpty() || currentSequenceId != startSequenceId) {
            return processRead(startSequenceId, maxSize);
        }

        // creating a pending read, and abort the previous one
        PendingRead oldPendingRead = pendingRead;
        pendingRead = new PendingRead(taskInstanceId, currentSequenceId, maxSize);

        if (oldPendingRead != null) {
            oldPendingRead.completeResultFutureWithEmpty();
        }

        return pendingRead.getResultFuture();
    }
@Test
    public void testSimplePendingRead()
    {
        SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

        // attempt to get a page
        ListenableFuture<BufferResult> future = buffer.get(BUFFER_ID, 0, sizeOfPages(2));
        assertFalse(future.isDone());

        // add three pages
        List<Page> pages = new LinkedList<>();
        for (int i = 0; i < 3; i++) {
            pages.add(createPage(i));
        }
        addPages(buffer, pages);

        // pending read should have two pages added
        assertBufferResultEquals(TYPES, getFuture(future, MAX_WAIT), bufferResult(0, createPage(0), createPage(1)));

        // checks we can still read first three pages
        assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 0, sizeOfPages(3), MAX_WAIT), createBufferResult(TASK_INSTANCE_ID, 0, pages));

        acknowledgeBufferResult(buffer, BUFFER_ID, 2);
        compareTotalBuffered(buffer, 3);

        assertBufferResultEquals(TYPES, getBufferResult(buffer, BUFFER_ID, 2, sizeOfPages(3), MAX_WAIT), bufferResult(2, createPage(2)));

        // file should be removed after acknowledging all three pages
        acknowledgeBufferResult(buffer, BUFFER_ID, 3);
        compareTotalBuffered(buffer, 0);

        // attempt to read, but nothing can be read
        future = buffer.get(BUFFER_ID, 3, sizeOfPages(3));
        assertFalse(future.isDone());
    }