ConsumerNetworkClient implements Closeable { public RequestFuture<ClientResponse> send(Node node, ApiKeys api, AbstractRequest request) { long now = time.milliseconds(); RequestFutureCompletionHandler future = new RequestFutureCompletionHandler(); RequestHeader header = client.nextRequestHeader(api); RequestSend send = new RequestSend(node.idString(), header, request.toStruct()); put(node, new ClientRequest(now, true, send, future)); return future; } ConsumerNetworkClient(KafkaClient client,
        Metadata metadata,
        Time time,
        long retryBackoffMs,
        long requestTimeoutMs); void schedule(DelayedTask task, long at); void unschedule(DelayedTask task); RequestFuture<ClientResponse> send(Node node,
        ApiKeys api,
        AbstractRequest request); Node leastLoadedNode(); void awaitMetadataUpdate(); void wakeup(); void ensureFreshMetadata(); void poll(RequestFuture<?> future); boolean poll(RequestFuture<?> future, long timeout); void poll(long timeout); void poll(long timeout, long now); void pollNoWakeup(); void executeDelayedTasks(long now); void awaitPendingRequests(Node node); int pendingRequestCount(Node node); int pendingRequestCount(); void disableWakeups(); void enableWakeups(); @Override void close(); boolean connectionFailed(Node node); void tryConnect(Node node);  }
@Test public void send() { client.prepareResponse(heartbeatResponse(Errors.NONE.code())); RequestFuture<ClientResponse> future = consumerClient.send(node, ApiKeys.METADATA, heartbeatRequest()); assertEquals(1, consumerClient.pendingRequestCount()); assertEquals(1, consumerClient.pendingRequestCount(node)); assertFalse(future.isDone()); consumerClient.poll(future); assertTrue(future.isDone()); assertTrue(future.succeeded()); ClientResponse clientResponse = future.value(); HeartbeatResponse response = new HeartbeatResponse(clientResponse.responseBody()); assertEquals(Errors.NONE.code(), response.errorCode()); }