NetworkClient implements KafkaClient { @Override public Node leastLoadedNode(long now) { List<Node> nodes = this.metadataUpdater.fetchNodes(); int inflight = Integer.MAX_VALUE; Node found = null; 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); int currInflight = this.inFlightRequests.inFlightRequestCount(node.idString()); if (currInflight == 0 && this.connectionStates.isConnected(node.idString())) { return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { inflight = currInflight; found = node; } } return found; } NetworkClient(Selectable selector,
        Metadata metadata,
        String clientId,
        int maxInFlightRequestsPerConnection,
        long reconnectBackoffMs,
        int socketSendBuffer,
        int socketReceiveBuffer,
        int requestTimeoutMs,
        Time time);  NetworkClient(Selectable selector,
        MetadataUpdater metadataUpdater,
        String clientId,
        int maxInFlightRequestsPerConnection,
        long reconnectBackoffMs,
        int socketSendBuffer,
        int socketReceiveBuffer,
        int requestTimeoutMs,
        Time time); private  NetworkClient(MetadataUpdater metadataUpdater,
        Metadata metadata,
        Selectable selector,
        String clientId,
        int maxInFlightRequestsPerConnection,
        long reconnectBackoffMs,
        int socketSendBuffer,
        int socketReceiveBuffer,
        int requestTimeoutMs,
        Time time); @Override boolean ready(Node node, long now); @Override void close(String nodeId); @Override long connectionDelay(Node node, long now); @Override boolean connectionFailed(Node node); @Override boolean isReady(Node node, long now); @Override void send(ClientRequest request, long now); @Override List<ClientResponse> poll(long timeout, long now); @Override int inFlightRequestCount(); @Override int inFlightRequestCount(String node); @Override RequestHeader nextRequestHeader(ApiKeys key); @Override RequestHeader nextRequestHeader(ApiKeys key, short version); @Override void wakeup(); @Override void close(); @Override Node leastLoadedNode(long now); static Struct parseResponse(ByteBuffer responseBuffer, RequestHeader requestHeader);  }
@Test public void testLeastLoadedNode() { Node leastNode = null; client.ready(node, time.milliseconds()); awaitReady(client, node); client.poll(1, time.milliseconds()); assertTrue("The client should be ready", client.isReady(node, time.milliseconds())); leastNode = client.leastLoadedNode(time.milliseconds()); assertEquals("There should be one leastloadednode", leastNode.id(), node.id()); time.sleep(reconnectBackoffMsTest); selector.close(node.idString()); client.poll(1, time.milliseconds()); assertFalse("After we forced the disconnection the client is no longer ready.", client.ready(node, time.milliseconds())); leastNode = client.leastLoadedNode(time.milliseconds()); assertEquals("There should be NO leastloadednode", leastNode, null); }