NetworkClient implements KafkaClient { @Override public void close(String nodeId) { selector.close(nodeId); for (ClientRequest request : inFlightRequests.clearAll(nodeId)) metadataUpdater.maybeHandleDisconnection(request); connectionStates.remove(nodeId); } 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 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 produceRequest = new ProduceRequest((short) 1, 1000, Collections.<TopicPartition, ByteBuffer>emptyMap()); RequestHeader reqHeader = client.nextRequestHeader(ApiKeys.PRODUCE); RequestSend send = new RequestSend(node.idString(), reqHeader, produceRequest.toStruct()); ClientRequest request = new ClientRequest(time.milliseconds(), true, send, null); client.send(request, time.milliseconds()); assertEquals("There should be 1 in-flight request after send", 1, client.inFlightRequestCount(node.idString())); client.close(node.idString()); assertEquals("There should be no in-flight request after close", 0, client.inFlightRequestCount(node.idString())); assertFalse("Connection should not be ready after close", client.isReady(node, 0)); }