@Test
    public void testPartialReceiveGracefulClose() throws Exception {
        String id = "0";
        blockingConnect(id);
        KafkaChannel channel = selector.channel(id);
        // Inject a NetworkReceive into Kafka channel with a large size
        injectNetworkReceive(channel, 100000);
        sendNoReceive(channel, 2); // Send some data that gets received as part of injected receive
        selector.poll(1000); // Wait until some data arrives, but not a completed receive
        assertEquals(0, selector.completedReceives().size());
        server.closeConnections();
        waitForCondition(() -> {
            try {
                selector.poll(100);
                return !selector.disconnected().isEmpty();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }, 10000, "Channel not disconnected");
        assertEquals(1, selector.disconnected().size());
        assertEquals(channel.id(), selector.disconnected().keySet().iterator().next());
        assertEquals(0, selector.completedReceives().size());
    }
Tests that graceful close is not delayed if only part of an incoming receive is
available in the socket buffer.
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java