@Test
    public void testConnectException() throws Exception {
        Metrics metrics = new Metrics();
        AtomicBoolean throwIOException = new AtomicBoolean();
        Selector selector = new ImmediatelyConnectingSelector(CONNECTION_MAX_IDLE_MS, metrics, time, "MetricGroup", channelBuilder, new LogContext()) {
            @Override
            protected SelectionKey registerChannel(String id, SocketChannel socketChannel, int interestedOps) throws IOException {
                SelectionKey key = super.registerChannel(id, socketChannel, interestedOps);
                key.cancel();
                if (throwIOException.get())
                    throw new IOException("Test exception");
                return key;
            }
        };

        try {
            verifyImmediatelyConnectedException(selector, "0");
            throwIOException.set(true);
            verifyImmediatelyConnectedException(selector, "1");
        } finally {
            selector.close();
            metrics.close();
        }
    }
Verify that if Selector#connect fails and throws an Exception, all related objects
are cleared immediately before the exception is propagated.
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java