@Test public void testClose() { client.ready(node, time.milliseconds()); awaitReady(client, node); client.poll(1, time.milliseconds()); assertTrue("The client should be ready", client.isReady(node, time.milliseconds())); ProduceRequest.Builder builder = new ProduceRequest.Builder(RecordBatch.CURRENT_MAGIC_VALUE, (short) 1, 1000, Collections.<TopicPartition, MemoryRecords>emptyMap()); ClientRequest request = client.newClientRequest(node.idString(), builder, time.milliseconds(), true); client.send(request, time.milliseconds()); assertEquals("There should be 1 in-flight request after send", 1, client.inFlightRequestCount(node.idString())); assertTrue(client.hasInFlightRequests(node.idString())); assertTrue(client.hasInFlightRequests()); client.close(node.idString()); assertEquals("There should be no in-flight request after close", 0, client.inFlightRequestCount(node.idString())); assertFalse(client.hasInFlightRequests(node.idString())); assertFalse(client.hasInFlightRequests()); assertFalse("Connection should not be ready after close", client.isReady(node, 0)); }
@Override public void close(String nodeId) { selector.close(nodeId); for (InFlightRequest request : inFlightRequests.clearAll(nodeId)) if (request.isInternalRequest && request.header.apiKey() == ApiKeys.METADATA.id) metadataUpdater.handleDisconnection(request.destination); connectionStates.remove(nodeId); }
NetworkClient implements KafkaClient { @Override public void close(String nodeId) { selector.close(nodeId); for (InFlightRequest request : inFlightRequests.clearAll(nodeId)) if (request.isInternalRequest && request.header.apiKey() == ApiKeys.METADATA.id) metadataUpdater.handleDisconnection(request.destination); connectionStates.remove(nodeId); } }
NetworkClient implements KafkaClient { @Override public void close(String nodeId) { selector.close(nodeId); for (InFlightRequest request : inFlightRequests.clearAll(nodeId)) if (request.isInternalRequest && request.header.apiKey() == ApiKeys.METADATA.id) metadataUpdater.handleDisconnection(request.destination); connectionStates.remove(nodeId); } 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 void close(String nodeId) { selector.close(nodeId); for (InFlightRequest request : inFlightRequests.clearAll(nodeId)) if (request.isInternalRequest && request.header.apiKey() == ApiKeys.METADATA.id) metadataUpdater.handleDisconnection(request.destination); connectionStates.remove(nodeId); } 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 void close(String nodeId) { selector.close(nodeId); for (InFlightRequest request : inFlightRequests.clearAll(nodeId)) if (request.isInternalRequest && request.header.apiKey() == ApiKeys.METADATA.id) metadataUpdater.handleDisconnection(request.destination); connectionStates.remove(nodeId); } 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();  }