Collection<ClientRequest> removeAllTimedOut(long now) {
            final List<ClientRequest> expiredRequests = new ArrayList<>();
            for (ArrayDeque<ClientRequest> requests : unsent.values()) {
                final Iterator<ClientRequest> requestIterator = requests.iterator();
                boolean foundExpiredRequest = false;
                while (requestIterator.hasNext() && !foundExpiredRequest) {
                    final ClientRequest request = requestIterator.next();
                    final long elapsedMs = Math.max(0, now - request.createdTimeMs());
                    if (elapsedMs > request.requestTimeoutMs()) {
                        expiredRequests.add(request);
                        requestIterator.remove();
                        foundExpiredRequest = true;
                    }
                }
            }
            return expiredRequests;
        }
An inter-broker send thread that utilizes a non-blocking network client.
server-common/src/main/java/org/apache/kafka/server/util/InterBrokerSendThread.java