@Override
    public LeastLoadedNode leastLoadedNode(long now) {
        List<Node> nodes = this.metadataUpdater.fetchNodes();
        if (nodes.isEmpty())
            throw new IllegalStateException("There are no nodes in the Kafka cluster");
        int inflight = Integer.MAX_VALUE;

        Node foundConnecting = null;
        Node foundCanConnect = null;
        Node foundReady = null;

        boolean atLeastOneConnectionReady = false;

        int offset = this.randOffset.nextInt(nodes.size());
        for (int i = 0; i < nodes.size(); i++) {
            int idx = (offset + i) % nodes.size();
            Node node = nodes.get(idx);

            if (!atLeastOneConnectionReady
                    && connectionStates.isReady(node.idString(), now)
                    && selector.isChannelReady(node.idString())) {
                atLeastOneConnectionReady = true;
            }

            if (canSendRequest(node.idString(), now)) {
                int currInflight = this.inFlightRequests.count(node.idString());
                if (currInflight == 0) {
                    // if we find an established connection with no in-flight requests we can stop right away
                    log.trace("Found least loaded node {} connected with no in-flight requests", node);
                    return new LeastLoadedNode(node, true);
                } else if (currInflight < inflight) {
                    // otherwise if this is the best we have found so far, record that
                    inflight = currInflight;
                    foundReady = node;
                }
            } else if (connectionStates.isPreparingConnection(node.idString())) {
                foundConnecting = node;
            } else if (canConnect(node, now)) {
                if (foundCanConnect == null ||
                        this.connectionStates.lastConnectAttemptMs(foundCanConnect.idString()) >
                                this.connectionStates.lastConnectAttemptMs(node.idString())) {
                    foundCanConnect = node;
                }
            } else {
                log.trace("Removing node {} from least loaded node selection since it is neither ready " +
                        "for sending or connecting", node);
            }
        }

        // We prefer established connections if possible. Otherwise, we will wait for connections
        // which are being established before connecting to new nodes.
        if (foundReady != null) {
            log.trace("Found least loaded node {} with {} inflight requests", foundReady, inflight);
            return new LeastLoadedNode(foundReady, atLeastOneConnectionReady);
        } else if (foundConnecting != null) {
            log.trace("Found least loaded connecting node {}", foundConnecting);
            return new LeastLoadedNode(foundConnecting, atLeastOneConnectionReady);
        } else if (foundCanConnect != null) {
            log.trace("Found least loaded node {} with no active connection", foundCanConnect);
            return new LeastLoadedNode(foundCanConnect, atLeastOneConnectionReady);
        } else {
            log.trace("Least loaded node selection failed to find an available node");
            return new LeastLoadedNode(null, atLeastOneConnectionReady);
        }
    }
@Test
    public void testLeastLoadedNode() {
        client.ready(node, time.milliseconds());
        assertFalse(client.isReady(node, time.milliseconds()));
        LeastLoadedNode leastLoadedNode = client.leastLoadedNode(time.milliseconds());
        assertEquals(node, leastLoadedNode.node());
        assertTrue(leastLoadedNode.hasNodeAvailableOrConnectionReady());

        awaitReady(client, node);
        client.poll(1, time.milliseconds());
        assertTrue(client.isReady(node, time.milliseconds()), "The client should be ready");

        // leastloadednode should be our single node
        leastLoadedNode = client.leastLoadedNode(time.milliseconds());
        assertTrue(leastLoadedNode.hasNodeAvailableOrConnectionReady());
        Node leastNode = leastLoadedNode.node();
        assertEquals(leastNode.id(), node.id(), "There should be one leastloadednode");

        // sleep for longer than reconnect backoff
        time.sleep(reconnectBackoffMsTest);

        // CLOSE node
        selector.serverDisconnect(node.idString());

        client.poll(1, time.milliseconds());
        assertFalse(client.ready(node, time.milliseconds()), "After we forced the disconnection the client is no longer ready.");
        leastLoadedNode = client.leastLoadedNode(time.milliseconds());
        assertFalse(leastLoadedNode.hasNodeAvailableOrConnectionReady());
        assertNull(leastLoadedNode.node(), "There should be NO leastloadednode");
    }