@Test
    public void testMetricsCleanupOnSelectorClose() throws Exception {
        Metrics metrics = new Metrics();
        Selector selector = new ImmediatelyConnectingSelector(CONNECTION_MAX_IDLE_MS, metrics, time, "MetricGroup", channelBuilder, new LogContext()) {
            @Override
            public void close(String id) {
                super.close(id);
                throw new RuntimeException();
            }
        };
        assertTrue(metrics.metrics().size() > 1);
        String id = "0";
        selector.connect(id, new InetSocketAddress("localhost", server.port), BUFFER_SIZE, BUFFER_SIZE);

        // Close the selector and ensure a RuntimeException has been throw
        assertThrows(RuntimeException.class, selector::close);

        // We should only have one remaining metric for kafka-metrics-count, which is a global metric
        assertEquals(1, metrics.metrics().size());
    }
Tests that a connect and disconnect in a single poll invocation results in the channel id being
in `disconnected`, but not `connected`.
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java