@Override
    public void send(ClientRequest request, long now) {
        if (!connectionState(request.destination()).isReady(now))
            throw new IllegalStateException("Cannot send " + request + " since the destination is not ready");

        // Check if the request is directed to a node with a pending authentication error.
        for (Iterator<Map.Entry<Node, Long>> authErrorIter =
             pendingAuthenticationErrors.entrySet().iterator(); authErrorIter.hasNext(); ) {
            Map.Entry<Node, Long> entry = authErrorIter.next();
            Node node = entry.getKey();
            long backoffMs = entry.getValue();
            if (node.idString().equals(request.destination())) {
                authErrorIter.remove();
                // Set up a disconnected ClientResponse and create an authentication error
                // for the affected node.
                authenticationFailed(node, backoffMs);
                AbstractRequest.Builder<?> builder = request.requestBuilder();
                short version = nodeApiVersions.latestUsableVersion(request.apiKey(), builder.oldestAllowedVersion(),
                    builder.latestAllowedVersion());
                ClientResponse resp = new ClientResponse(request.makeHeader(version), request.callback(), request.destination(),
                    request.createdTimeMs(), time.milliseconds(), true, null,
                        new AuthenticationException("Authentication failed"), null);
                responses.add(resp);
                return;
            }
        }
        Iterator<FutureResponse> iterator = futureResponses.iterator();
        while (iterator.hasNext()) {
            FutureResponse futureResp = iterator.next();
            if (futureResp.node != null && !request.destination().equals(futureResp.node.idString()))
                continue;

            AbstractRequest.Builder<?> builder = request.requestBuilder();

            try {
                short version = nodeApiVersions.latestUsableVersion(request.apiKey(), builder.oldestAllowedVersion(),
                        builder.latestAllowedVersion());

                UnsupportedVersionException unsupportedVersionException = null;
                if (futureResp.isUnsupportedRequest) {
                    unsupportedVersionException = new UnsupportedVersionException(
                            "Api " + request.apiKey() + " with version " + version);
                } else {
                    AbstractRequest abstractRequest = request.requestBuilder().build(version);
                    if (!futureResp.requestMatcher.matches(abstractRequest))
                        throw new IllegalStateException("Request matcher did not match next-in-line request "
                                + abstractRequest + " with prepared response " + futureResp.responseBody);
                }

                ClientResponse resp = new ClientResponse(request.makeHeader(version), request.callback(), request.destination(),
                        request.createdTimeMs(), time.milliseconds(), futureResp.disconnected,
                        unsupportedVersionException, null, futureResp.responseBody);
                responses.add(resp);
            } catch (UnsupportedVersionException unsupportedVersionException) {
                ClientResponse resp = new ClientResponse(request.makeHeader(builder.latestAllowedVersion()), request.callback(), request.destination(),
                        request.createdTimeMs(), time.milliseconds(), false, unsupportedVersionException, null, null);
                responses.add(resp);
            }
            iterator.remove();
            return;
        }

        this.requests.add(request);
    }
A mock network client for use testing code
clients/src/test/java/org/apache/kafka/clients/MockClient.java