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


##########
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 could still be simplified a lot like I was suggesting in the comment 
above. Not handling all errors, only the fencing/fail ones. For all the rest is 
a common action that could be done with a single 
`nonRetriableErrorHandler.handle(error.get().exception());`



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