@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 testMultiplePendingReads()
    {
        SpoolingOutputBuffer buffer = createSpoolingOutputBuffer();

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

        ListenableFuture<BufferResult> newPendingRead = buffer.get(BUFFER_ID, 0, sizeOfPages(3));
        assertFalse(newPendingRead.isDone());
        assertTrue(oldPendingRead.isDone());

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

        assertBufferResultEquals(TYPES, getFuture(oldPendingRead, MAX_WAIT), emptyResults(TASK_INSTANCE_ID, 0, false));
        assertBufferResultEquals(TYPES, getFuture(newPendingRead, MAX_WAIT), createBufferResult(TASK_INSTANCE_ID, 0, pages));
    }