@Test
    public void testConnectDisconnectDuringInSinglePoll() throws Exception {
        // channel is connected, not ready and it throws an exception during prepare
        KafkaChannel kafkaChannel = mock(KafkaChannel.class);
        when(kafkaChannel.id()).thenReturn("1");
        when(kafkaChannel.socketDescription()).thenReturn("");
        when(kafkaChannel.state()).thenReturn(ChannelState.NOT_CONNECTED);
        when(kafkaChannel.finishConnect()).thenReturn(true);
        when(kafkaChannel.isConnected()).thenReturn(true);
        when(kafkaChannel.ready()).thenReturn(false);
        doThrow(new IOException()).when(kafkaChannel).prepare();

        SelectionKey selectionKey = mock(SelectionKey.class);
        when(kafkaChannel.selectionKey()).thenReturn(selectionKey);
        SocketChannel socket = SocketChannel.open();
        when(selectionKey.channel()).thenReturn(socket);
        when(selectionKey.readyOps()).thenReturn(SelectionKey.OP_CONNECT);
        when(selectionKey.attachment()).thenReturn(kafkaChannel);

        Set<SelectionKey> selectionKeys = Utils.mkSet(selectionKey);
        selector.pollSelectionKeys(selectionKeys, false, System.nanoTime());

        assertFalse(selector.connected().contains(kafkaChannel.id()));
        assertTrue(selector.disconnected().containsKey(kafkaChannel.id()));

        verify(kafkaChannel, atLeastOnce()).ready();
        verify(kafkaChannel).disconnect();
        verify(kafkaChannel).close();
        verify(selectionKey).cancel();
        socket.close();
    }
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