@Override
    public synchronized void close()
    {
        if (!closed.compareAndSet(false, true)) {
            return;
        }

        for (PageBufferClient client : allClients.values()) {
            closeQuietly(client);
        }
        pageBuffer.clear();
        systemMemoryContext.setBytes(0);
        bufferRetainedSizeInBytes = 0;
        if (pageBuffer.peekLast() != NO_MORE_PAGES) {
            checkState(pageBuffer.add(NO_MORE_PAGES), "Could not add no more pages marker");
        }
        notifyBlockedCallers();
    }
@Test
    public void testClose()
            throws Exception
    {
        DataSize bufferCapacity = new DataSize(1, BYTE);
        DataSize maxResponseSize = new DataSize(1, BYTE);
        MockExchangeRequestProcessor processor = new MockExchangeRequestProcessor(maxResponseSize);

        URI location = URI.create("http://localhost:8080");
        processor.addPage(location, createPage(1));
        processor.addPage(location, createPage(2));
        processor.addPage(location, createPage(3));

        ExchangeClient exchangeClient = createExchangeClient(processor, bufferCapacity, maxResponseSize);

        exchangeClient.addLocation(location, TaskId.valueOf("taskid.0.0.0.0"));
        exchangeClient.noMoreLocations();

        // fetch a page
        assertFalse(exchangeClient.isClosed());
        assertPageEquals(getNextPage(exchangeClient), createPage(1));

        // close client while pages are still available
        exchangeClient.close();
        waitUntilEquals(exchangeClient::isFinished, true, new Duration(5, SECONDS));
        assertTrue(exchangeClient.isClosed());
        assertNull(exchangeClient.pollPage());
        assertEquals(exchangeClient.getStatus().getBufferedPages(), 0);
        assertEquals(exchangeClient.getStatus().getBufferedBytes(), 0);

        // client should have sent only 2 requests: one to get all pages and once to get the done signal
        Optional<PageBufferClientStatus> clientStatusOptional = exchangeClient.getStatus().getPageBufferClientStatuses().stream().filter(pageBufferClientStatus -> pageBufferClientStatus.getUri().equals(location)).findFirst();
        assertTrue(clientStatusOptional.isPresent());
        assertStatus(clientStatusOptional.get(), "closed", "not scheduled");
    }