public void acknowledgePages(long sequenceId)
    {
        checkArgument(sequenceId >= 0, "Invalid sequence id");
        // Fast path early-return without synchronizing
        if (destroyed.get() || sequenceId < currentSequenceId.get()) {
            return;
        }

        ImmutableList.Builder<SerializedPageReference> removedPages;
        synchronized (this) {
            if (destroyed.get()) {
                return;
            }

            // if pages have already been acknowledged, just ignore this
            long oldCurrentSequenceId = currentSequenceId.get();
            if (sequenceId < oldCurrentSequenceId) {
                return;
            }

            int pagesToRemove = toIntExact(sequenceId - oldCurrentSequenceId);
            checkArgument(pagesToRemove <= pages.size(), "Invalid sequence id");
            removedPages = ImmutableList.builderWithExpectedSize(pagesToRemove);

            long bytesRemoved = 0;
            for (int i = 0; i < pagesToRemove; i++) {
                SerializedPageReference removedPage = pages.removeFirst();
                removedPages.add(removedPage);
                bytesRemoved += removedPage.getRetainedSizeInBytes();
            }

            // update current sequence id
            verify(currentSequenceId.compareAndSet(oldCurrentSequenceId, oldCurrentSequenceId + pagesToRemove));

            // update memory tracking
            verify(bufferedBytes.addAndGet(-bytesRemoved) >= 0);
        }
        // dereference outside of synchronized to avoid making a callback while holding a lock
        dereferencePages(removedPages.build(), onPagesReleased);
    }
@Test
    public void testBufferResults()
    {
        ClientBuffer buffer = new ClientBuffer(TASK_INSTANCE_ID, BUFFER_ID, NOOP_RELEASE_LISTENER);

        long totalSizeOfPagesInBytes = 0;
        for (int i = 0; i < 3; i++) {
            Page page = createPage(i);
            SerializedPageReference pageReference = new SerializedPageReference(PAGES_SERDE.serialize(page), 1, Lifespan.taskWide());
            totalSizeOfPagesInBytes = totalSizeOfPagesInBytes + pageReference.getSerializedPage().getRetainedSizeInBytes();
            addPage(buffer, page);
        }
        // Everything buffered
        assertBufferInfo(buffer, 3, 0, totalSizeOfPagesInBytes);

        BufferResult bufferResult = getBufferResult(buffer, 0, sizeOfPages(1), NO_WAIT);
        long remainingBytes = totalSizeOfPagesInBytes - bufferResult.getBufferedBytes();
        assertEquals(bufferResult.getBufferedBytes(), sizeOfPages(1).toBytes());
        assertBufferInfo(buffer, 3, 0, totalSizeOfPagesInBytes);

        buffer.acknowledgePages(bufferResult.getNextToken());
        assertBufferInfo(buffer, 2, 1, remainingBytes);
    }