### Question:
ExchangeClient implements Closeable { @Override public synchronized void close() { if (!closed.compareAndSet(false, true)) { return; } for (HttpPageBufferClient 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(); } ExchangeClient(
            String selfAddress,
            DataIntegrityVerification dataIntegrityVerification,
            DataSize bufferCapacity,
            DataSize maxResponseSize,
            int concurrentRequestMultiplier,
            Duration maxErrorDuration,
            boolean acknowledgePages,
            HttpClient httpClient,
            ScheduledExecutorService scheduler,
            LocalMemoryContext systemMemoryContext,
            Executor pageBufferClientCallbackExecutor); ExchangeClientStatus getStatus(); synchronized void addLocation(URI location); synchronized void noMoreLocations(); WorkProcessor<SerializedPage> pages(); @Nullable SerializedPage pollPage(); boolean isFinished(); boolean isClosed(); @Override synchronized void close(); synchronized void scheduleRequestIfNecessary(); synchronized ListenableFuture<?> isBlocked();  }### Answer:
@Test public void testClose() throws Exception { DataSize maxResponseSize = DataSize.ofBytes(1); MockExchangeRequestProcessor processor = new MockExchangeRequestProcessor(maxResponseSize); URI location = URI.create("http: processor.addPage(location, createPage(1)); processor.addPage(location, createPage(2)); processor.addPage(location, createPage(3)); @SuppressWarnings("resource") ExchangeClient exchangeClient = new ExchangeClient( "localhost", DataIntegrityVerification.ABORT, DataSize.ofBytes(1), maxResponseSize, 1, new Duration(1, TimeUnit.MINUTES), true, new TestingHttpClient(processor, newCachedThreadPool(daemonThreadsNamed(getClass().getSimpleName() + "-testClose-%s"))), scheduler, new SimpleLocalMemoryContext(newSimpleAggregatedMemoryContext(), "test"), pageBufferClientCallbackExecutor); exchangeClient.addLocation(location); exchangeClient.noMoreLocations(); assertEquals(exchangeClient.isClosed(), false); assertPageEquals(getNextPage(exchangeClient), createPage(1)); exchangeClient.close(); while (!exchangeClient.isFinished()) { MILLISECONDS.sleep(10); } assertEquals(exchangeClient.isClosed(), true); assertNull(exchangeClient.pollPage()); assertEquals(exchangeClient.getStatus().getBufferedPages(), 0); assertEquals(exchangeClient.getStatus().getBufferedBytes(), 0); PageBufferClientStatus clientStatus = exchangeClient.getStatus().getPageBufferClientStatuses().get(0); assertEquals(clientStatus.getUri(), location); assertEquals(clientStatus.getState(), "closed", "status"); assertEquals(clientStatus.getHttpRequestState(), "not scheduled", "httpRequestState"); }