private void checkDisconnects(long now) {
        // any disconnects affecting requests that have already been transmitted will be handled
        // by NetworkClient, so we just need to check whether connections for any of the unsent
        // requests have been disconnected; if they have, then we complete the corresponding future
        // and set the disconnect flag in the ClientResponse
        final Iterator<Map.Entry<Node, ArrayDeque<ClientRequest>>> iterator = unsentRequests.iterator();
        while (iterator.hasNext()) {
            final Map.Entry<Node, ArrayDeque<ClientRequest>> entry = iterator.next();
            final Node node = entry.getKey();
            final ArrayDeque<ClientRequest> requests = entry.getValue();
            if (!requests.isEmpty() && networkClient.connectionFailed(node)) {
                iterator.remove();
                for (ClientRequest request : requests) {
                    final AuthenticationException authenticationException = networkClient.authenticationException(node);
                    if (authenticationException != null) {
                        log.error("Failed to send the following request due to authentication error: {}", request);
                    }
                    completeWithDisconnect(request, now, authenticationException);
                }
            }
        }
    }
An inter-broker send thread that utilizes a non-blocking network client.
server-common/src/main/java/org/apache/kafka/server/util/InterBrokerSendThread.java