[KAFKA-15534] Propagate client response time when timeout to the request handler

• Type: Sub-task
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Fix Version/s: 3.7.0
• Component/s: clients, consumer
• Labels: consumer-threading-refactor kip-848-client-support kip-848-e2e kip-848-preview
• Assignee: Philip Nee
• Reporter: Philip Nee
• Created: 03/Oct/23 18:59
• Updated: 19/Dec/23 03:15
• Resolved: 02/Nov/23 16:33

Description

Currently, we don't have a good way to propagate the response time to the handler when timeout is thrown.

unsent.handler.onFailure(new TimeoutException(
    "Failed to send request after " + unsent.timer.timeoutMs() + " ms.")); 

The current request manager invoke a system call to retrieve the response time, which is not idea because it is already available at network client

This is an example of the coordinator request manager:

unsentRequest.future().whenComplete((clientResponse, throwable) -> {
    long responseTimeMs = time.milliseconds();
    if (clientResponse != null) {
        FindCoordinatorResponse response = (FindCoordinatorResponse) clientResponse.responseBody();
        onResponse(responseTimeMs, response);
    } else {
        onFailedResponse(responseTimeMs, throwable);
    }
}); 

But in the networkClientDelegate, we should utilize the currentTimeMs in the trySend to avoid calling time.milliseconds():

private void trySend(final long currentTimeMs) {
...
unsent.handler.onFailure(new TimeoutException(
        "Failed to send request after " + unsent.timer.timeoutMs() + " ms."));
    continue;
}
} 

Issue Links

• links to: GitHub Pull Request #14532