@Test public void testLeastLoadedNode() { client.ready(node, time.milliseconds()); awaitReady(client, node); client.poll(1, time.milliseconds()); assertTrue("The client should be ready", client.isReady(node, time.milliseconds())); Node 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); }
@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.count(node.idString()); if (currInflight == 0 && isReady(node, now)) { log.trace("Found least loaded node {} connected with no in-flight requests", node); return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { inflight = currInflight; found = node; } else if (log.isTraceEnabled()) { log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight); } } if (found != null) log.trace("Found least loaded node {}", found); else log.trace("Least loaded node selection failed to find an available node"); return found; }
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.count(node.idString()); if (currInflight == 0 && isReady(node, now)) { log.trace("Found least loaded node {} connected with no in-flight requests", node); return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { inflight = currInflight; found = node; } else if (log.isTraceEnabled()) { log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight); } } if (found != null) log.trace("Found least loaded node {}", found); else log.trace("Least loaded node selection failed to find an available node"); return found; } }
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.count(node.idString()); if (currInflight == 0 && isReady(node, now)) { log.trace("Found least loaded node {} connected with no in-flight requests", node); return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { inflight = currInflight; found = node; } else if (log.isTraceEnabled()) { log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight); } } if (found != null) log.trace("Found least loaded node {}", found); else log.trace("Least loaded node selection failed to find an available node"); return found; } NetworkClient(Selectable selector,
                         Metadata metadata,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int requestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions);  NetworkClient(Selectable selector,
            Metadata metadata,
            String clientId,
            int maxInFlightRequestsPerConnection,
            long reconnectBackoffMs,
            long reconnectBackoffMax,
            int socketSendBuffer,
            int socketReceiveBuffer,
            int requestTimeoutMs,
            Time time,
            boolean discoverBrokerVersions,
            ApiVersions apiVersions,
            Sensor throttleTimeSensor);  NetworkClient(Selectable selector,
                         MetadataUpdater metadataUpdater,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int requestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions); private  NetworkClient(MetadataUpdater metadataUpdater,
                          Metadata metadata,
                          Selectable selector,
                          String clientId,
                          int maxInFlightRequestsPerConnection,
                          long reconnectBackoffMs,
                          long reconnectBackoffMax,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int requestTimeoutMs,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
                          Sensor throttleTimeSensor); }
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.count(node.idString()); if (currInflight == 0 && isReady(node, now)) { log.trace("Found least loaded node {} connected with no in-flight requests", node); return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { inflight = currInflight; found = node; } else if (log.isTraceEnabled()) { log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight); } } if (found != null) log.trace("Found least loaded node {}", found); else log.trace("Least loaded node selection failed to find an available node"); return found; } NetworkClient(Selectable selector,
                         Metadata metadata,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int requestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions);  NetworkClient(Selectable selector,
            Metadata metadata,
            String clientId,
            int maxInFlightRequestsPerConnection,
            long reconnectBackoffMs,
            long reconnectBackoffMax,
            int socketSendBuffer,
            int socketReceiveBuffer,
            int requestTimeoutMs,
            Time time,
            boolean discoverBrokerVersions,
            ApiVersions apiVersions,
            Sensor throttleTimeSensor);  NetworkClient(Selectable selector,
                         MetadataUpdater metadataUpdater,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int requestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions); private  NetworkClient(MetadataUpdater metadataUpdater,
                          Metadata metadata,
                          Selectable selector,
                          String clientId,
                          int maxInFlightRequestsPerConnection,
                          long reconnectBackoffMs,
                          long reconnectBackoffMax,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int requestTimeoutMs,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
                          Sensor throttleTimeSensor); @Override // 检测是否可以向一个Node发送请求 boolean ready(Node node, long now); @Override void disconnect(String nodeId); @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 // 调用Selector.poll 进行网络IO // 执行完成后 触发回调逻辑 List<ClientResponse> poll(long timeout, long now); @Override int inFlightRequestCount(); @Override boolean hasInFlightRequests(); @Override int inFlightRequestCount(String node); @Override boolean hasInFlightRequests(String node); @Override boolean hasReadyNodes(); @Override void wakeup(); @Override void close(); @Override Node leastLoadedNode(long now); static AbstractResponse parseResponse(ByteBuffer responseBuffer, RequestHeader requestHeader); @Override ClientRequest newClientRequest(String nodeId, AbstractRequest.Builder<?> requestBuilder, long createdTimeMs,
                                          boolean expectResponse); @Override ClientRequest newClientRequest(String nodeId, AbstractRequest.Builder<?> requestBuilder, long createdTimeMs,
                                          boolean expectResponse, RequestCompletionHandler callback); boolean discoverBrokerVersions(); }
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.count(node.idString()); if (currInflight == 0 && isReady(node, now)) { log.trace("Found least loaded node {} connected with no in-flight requests", node); return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { inflight = currInflight; found = node; } else if (log.isTraceEnabled()) { log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight); } } if (found != null) log.trace("Found least loaded node {}", found); else log.trace("Least loaded node selection failed to find an available node"); return found; } NetworkClient(Selectable selector,
                         Metadata metadata,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int requestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions);  NetworkClient(Selectable selector,
            Metadata metadata,
            String clientId,
            int maxInFlightRequestsPerConnection,
            long reconnectBackoffMs,
            long reconnectBackoffMax,
            int socketSendBuffer,
            int socketReceiveBuffer,
            int requestTimeoutMs,
            Time time,
            boolean discoverBrokerVersions,
            ApiVersions apiVersions,
            Sensor throttleTimeSensor);  NetworkClient(Selectable selector,
                         MetadataUpdater metadataUpdater,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int requestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions); private  NetworkClient(MetadataUpdater metadataUpdater,
                          Metadata metadata,
                          Selectable selector,
                          String clientId,
                          int maxInFlightRequestsPerConnection,
                          long reconnectBackoffMs,
                          long reconnectBackoffMax,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int requestTimeoutMs,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
                          Sensor throttleTimeSensor); @Override // 检测是否可以向一个Node发送请求 boolean ready(Node node, long now); @Override void disconnect(String nodeId); @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 // 调用Selector.poll 进行网络IO // 执行完成后 触发回调逻辑 List<ClientResponse> poll(long timeout, long now); @Override int inFlightRequestCount(); @Override boolean hasInFlightRequests(); @Override int inFlightRequestCount(String node); @Override boolean hasInFlightRequests(String node); @Override boolean hasReadyNodes(); @Override void wakeup(); @Override void close(); @Override Node leastLoadedNode(long now); static AbstractResponse parseResponse(ByteBuffer responseBuffer, RequestHeader requestHeader); @Override ClientRequest newClientRequest(String nodeId, AbstractRequest.Builder<?> requestBuilder, long createdTimeMs,
                                          boolean expectResponse); @Override ClientRequest newClientRequest(String nodeId, AbstractRequest.Builder<?> requestBuilder, long createdTimeMs,
                                          boolean expectResponse, RequestCompletionHandler callback); boolean discoverBrokerVersions();  }