private void verifyChannelExpiry(KafkaChannel channel) throws Exception {
        String id = channel.id();
        selector.mute(id); // Mute to allow channel to be expired even if more data is available for read
        time.sleep(CONNECTION_MAX_IDLE_MS + 1_000);
        selector.poll(0);
        assertNull(selector.channel(id), "Channel not expired");
        assertNull(selector.closingChannel(id), "Channel not removed from closingChannels");
        assertEquals(ChannelState.EXPIRED, channel.state());
        assertNull(channel.selectionKey().attachment());
        assertTrue(selector.disconnected().containsKey(id), "Disconnect not notified");
        assertEquals(ChannelState.EXPIRED, selector.disconnected().get(id));
        verifySelectorEmpty();
    }
Verifies that a muted connection closed by peer is expired on idle timeout even if there are pending
receives on the socket.
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java