@Test
    public void testGracefulClose() throws Exception {
        int maxReceiveCountAfterClose = 0;
        for (int i = 6; i <= 100 && maxReceiveCountAfterClose < 5; i++) {
            int receiveCount = 0;
            KafkaChannel channel = createConnectionWithPendingReceives(i);
            // Poll until one or more receives complete and then close the server-side connection
            waitForCondition(() -> {
                selector.poll(1000);
                return !selector.completedReceives().isEmpty();
            }, 5000, "Receive not completed");
            server.closeConnections();
            while (selector.disconnected().isEmpty()) {
                selector.poll(1);
                receiveCount += selector.completedReceives().size();
                assertTrue(selector.completedReceives().size() <= 1, "Too many completed receives in one poll");
            }
            assertEquals(channel.id(), selector.disconnected().keySet().iterator().next());
            maxReceiveCountAfterClose = Math.max(maxReceiveCountAfterClose, receiveCount);
        }
        assertTrue(maxReceiveCountAfterClose >= 5, "Too few receives after close: " + maxReceiveCountAfterClose);
    }
Tests that graceful close of channel processes remaining data from socket read buffers.
Since we cannot determine how much data is available in the buffers, this test verifies that
multiple receives are completed after server shuts down connections, with retries to tolerate
cases where data may not be available in the socket buffer.
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java