philipnee commented on code in PR #14532:
URL: https://github.com/apache/kafka/pull/14532#discussion_r1358656536


##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -184,13 +184,11 @@ private NetworkClientDelegate.UnsentRequest 
makeHeartbeatRequest() {
         NetworkClientDelegate.UnsentRequest request = new 
NetworkClientDelegate.UnsentRequest(
             new ConsumerGroupHeartbeatRequest.Builder(data),
             coordinatorRequestManager.coordinator());
-        request.future().whenComplete((response, exception) -> {
+        request.handler().whenComplete((response, exception) -> {
             if (response != null) {
                 onResponse((ConsumerGroupHeartbeatResponse) 
response.responseBody(), response.receivedTimeMs());
             } else {
-                // TODO: Currently, we lack a good way to propage the response 
time from the network client to the
-                //  request handler. We will need to store the response time 
in the handler to make it accessible.
-                onFailure(exception, time.milliseconds());
+                onFailure(exception, request.handler().completionTimeMs());

Review Comment:
   I think the only time we invoke onComplete() is when a response is available 
(see below). If the response is `null`, which indicates the request has not 
been sent out; therefore, in the `NetworkClientDelegate`, we need to actively 
fail the request on timeout.
   
   onComplete when response is available:
   ```
   private void completeResponses(List<ClientResponse> responses) {
           for (ClientResponse response : responses) {
               try {
                   response.onComplete();
               } catch (Exception e) {
                   log.error("Uncaught error in request completion:", e);
               }
           }
       }
   ```
   
   We actively expires the unsent request and fail them with TimeoutException:
   ```
   if (unsent.timer.isExpired()) {
                   iterator.remove();
                   unsent.handler.onFailure(currentTimeMs, new TimeoutException(
                       "Failed to send request after " + 
unsent.timer.timeoutMs() + " ms."));
                   continue;
               }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Reply via email to