[GitHub] [kafka] lianetm commented on a diff in pull request #14364: KAFKA-15278: Implement HeartbeatRequestManager to handle heartbeat requests

[via GitHub]

lianetm commented on code in PR #14364:
URL: https://github.com/apache/kafka/pull/14364#discussion_r1331811249


##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -158,27 +165,27 @@ private void onSuccess(final 
ConsumerGroupHeartbeatResponse response, long curre
             return;
         }
 
-        onError(response, currentTimeMs);
+        onErrorResponse(response, currentTimeMs);
     }
 
-    private void onError(final ConsumerGroupHeartbeatResponse response,
-                         final long currentTimeMs) {
-
+    private void onErrorResponse(final ConsumerGroupHeartbeatResponse response,
+                                 final long currentTimeMs) {
         this.heartbeatRequestState.onFailedAttempt(currentTimeMs);
         short errorCode = response.data().errorCode();
         if (errorCode == Errors.NOT_COORDINATOR.code() || errorCode == 
Errors.COORDINATOR_NOT_AVAILABLE.code()) {
-            logger.info("GroupHeartbeatRequest failed: coordinator is either 
not started or not valid. Retrying in " +
-                    "{}ms: {}",
-                heartbeatRequestState.remainingBackoffMs(currentTimeMs),
-                response.data().errorMessage());
+            logInfo("Coordinator is either not started or not valid. 
Retrying", response, currentTimeMs);
             
coordinatorRequestManager.markCoordinatorUnknown(response.data().errorMessage(),
 currentTimeMs);
         } else if (errorCode == Errors.COORDINATOR_LOAD_IN_PROGRESS.code()) {
             // retry
-            logger.info("GroupHeartbeatRequest failed: Coordinator {} is 
loading. Retrying in {}ms: {}",
-                coordinatorRequestManager.coordinator(),
-                heartbeatRequestState.remainingBackoffMs(currentTimeMs),
-                response.data().errorMessage());
-        } else if (errorCode == Errors.GROUP_AUTHORIZATION_FAILED.code()) {
+            logInfo("Coordinator {} is loading. Retrying", response, 
currentTimeMs);
+        } else {
+            onFatalErrorResponse(response);
+        }
+    }
+
+    private void onFatalErrorResponse(final ConsumerGroupHeartbeatResponse 
response) {

Review Comment:
   This whole func would completely disappear if we agree on the something like 
the draft PR [here](https://github.com/apache/kafka/pull/14413)



-- 
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