@Override
    public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException {
        ensureNotRegistered(id);
        SocketChannel socketChannel = SocketChannel.open();
        SelectionKey key = null;
        try {
            configureSocketChannel(socketChannel, sendBufferSize, receiveBufferSize);
            boolean connected = doConnect(socketChannel, address);
            key = registerChannel(id, socketChannel, SelectionKey.OP_CONNECT);

            if (connected) {
                // OP_CONNECT won't trigger for immediately connected channels
                log.debug("Immediately connected to node {}", id);
                immediatelyConnectedKeys.add(key);
                key.interestOps(0);
            }
        } catch (IOException | RuntimeException e) {
            if (key != null)
                immediatelyConnectedKeys.remove(key);
            channels.remove(id);
            socketChannel.close();
            throw e;
        }
    }
Begin connecting to the given address and add the connection to this nioSelector associated with the given id
number.
<p>
Note that this call only initiates the connection, which will be completed on a future {@link #poll(long)}
call. Check {@link #connected()} to see which (if any) connections have completed after a given poll call.
@param id The id for the new connection
@param address The address to connect to
@param sendBufferSize The send buffer for the new connection
@param receiveBufferSize The receive buffer for the new connection
@throws IllegalStateException if there is already a connection for that id
@throws IOException if DNS resolution fails on the hostname or if the broker is down
clients/src/main/java/org/apache/kafka/common/network/Selector.java