vamossagar12 commented on code in PR #12561:
URL: https://github.com/apache/kafka/pull/12561#discussion_r962289115
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java:
##########
@@ -307,9 +309,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++;
+ delay = (int)
consecutiveRevokingRebalancesBackoff.backoff(numSuccessiveRevokingRebalances);
+ log.debug("Consecutive revoking rebalances observed. Need to
wait for {} ms", delay);
Review Comment:
Thanks @C0urante , @yashmayya I made the change as suggested. I had made
missed the fact that since the workers would join immediately post a
load-balancing rebalance, introducing delay for that round is needless. I
needed to tweak some logic at a few places so a review would certainly be
helpful!
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
