@Test
    public void testCloseOldestConnectionWithMultiplePendingReceives() throws Exception {
        int expectedReceives = 5;
        KafkaChannel channel = createConnectionWithPendingReceives(expectedReceives);
        int completedReceives = selector.completedReceives().size();

        while (selector.disconnected().isEmpty()) {
            time.sleep(CONNECTION_MAX_IDLE_MS + 1_000);
            selector.poll(completedReceives == expectedReceives ? 0 : 1_000);
            completedReceives += selector.completedReceives().size();
        }

        assertEquals(expectedReceives, completedReceives);
        assertNull(selector.channel(channel.id()), "Channel not expired");
        assertNull(selector.closingChannel(channel.id()), "Channel not expired");
        assertTrue(selector.disconnected().containsKey(channel.id()), "Disconnect not notified");
        assertTrue(selector.completedReceives().isEmpty(), "Unexpected receive");
    }
Verifies that sockets with incoming data available are not expired.
For PLAINTEXT, pending receives are always read from socket without any buffering, so this
test is only verifying that channels are not expired while there is data to read from socket.
For SSL, pending receives may also be in SSL netReadBuffer or appReadBuffer. So the test verifies
that connection is not expired when data is available from buffers or network.
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java