private void timeoutCallsInFlight(TimeoutProcessor processor) {
            int numTimedOut = 0;
            for (Map.Entry<String, Call> entry : callsInFlight.entrySet()) {
                Call call = entry.getValue();
                String nodeId = entry.getKey();
                if (processor.callHasExpired(call)) {
                    log.info("Disconnecting from {} due to timeout while awaiting {}", nodeId, call);
                    client.disconnect(nodeId);
                    numTimedOut++;
                    // We don't remove anything from the callsInFlight data structure. Because the connection
                    // has been closed, the calls should be returned by the next client#poll(),
                    // and handled at that point.
                }
            }
            if (numTimedOut > 0)
                log.debug("Timed out {} call(s) in flight.", numTimedOut);
        }
Time out expired calls that are in flight.

Calls that are in flight may have been partially or completely sent over the wire. They may
even be in the process of being processed by the remote server. At the moment, our only option
to time them out is to close the entire connection.

@param processor         The timeout processor.
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java