[GitHub] [kafka] showuon commented on a diff in pull request #12561: KAFKA-12495: Exponential backoff retry to prevent rebalance storms when worker joins after revoking rebalance

[GitBox <gi...@apache.org>]

showuon commented on code in PR #12561:
URL: https://github.com/apache/kafka/pull/12561#discussion_r956850067


##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java:
##########
@@ -584,9 +606,18 @@ private Map<String, ConnectorsAndTasks> performTaskRevocation(ConnectorsAndTasks
         Collection<WorkerLoad> existingWorkers = completeWorkerAssignment.stream()
                 .filter(wl -> wl.size() > 0)
                 .collect(Collectors.toList());
-        int existingWorkersNum = existingWorkers.size();
+
+        // existingWorkers => workers with load > 0.
+        // if existingWorkers is empty, no workers have any load.
+        int existingWorkersNum = existingWorkers.size(); // count of workers having non-zero load.
         int totalWorkersNum = completeWorkerAssignment.size();
-        int newWorkersNum = totalWorkersNum - existingWorkersNum;
+        int newWorkersNum = totalWorkersNum - existingWorkersNum; // <total workers - workers with non-0 load>
+        // if existingWorkersNum == 0, no workers have any load i.e all workers are new.
+        // newWorkersNum = 0 when all workers have non-zero load.

Review Comment:
   newWorkersNum =[=] 0 ?



##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java:
##########
@@ -285,18 +292,11 @@ ClusterAssignment performTaskAssignment(
 
         handleLostAssignments(lostAssignments, newSubmissions, completeWorkerAssignment);
 
+        log.debug("Can leader revoke tasks in this assignment? {} (delay: {})", delay == 0, delay);

Review Comment:
   Since we don't have `canRevoke` now, I think we should rephrase this sentence. ex:
   ```java
   if (delay > 0)  
       log.debug("Delaying {}ms for revoking tasks.");
   ```
   WDYT?



##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java:
##########
@@ -307,9 +307,31 @@ ClusterAssignment performTaskAssignment(
                     existing.tasks().addAll(assignment.tasks());
                 }
             );
-            canRevoke = toExplicitlyRevoke.size() == 0;
+
+            // If this round and the previous round involved revocation, we will do an exponential
+            // backoff delay to prevent rebalance storms.
+            if (revokedInPrevious && !toExplicitlyRevoke.isEmpty()) {
+                numSuccessiveRevokingRebalances++;
+                long processAfter = consecutiveRevokingRebalancesBackoff.backoff(numSuccessiveRevokingRebalances);
+                log.debug("Consecutive revoking rebalances observed. Need to wait for {} ms", processAfter);
+                time.sleep(processAfter); // Is it a good idea to sleep?

Review Comment:
   No, we can't sleep during the rebalance process. If there are some new tasks/connectors needed to distribute to workers, it'll be delayed for `processAfter` ms. I think we should use the existing `delay`/`scheduledRebalance` mechanism to delay the revocation. 



-- 
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-unsubscribe@kafka.apache.org

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