vamossagar12 commented on code in PR #12561:
URL: https://github.com/apache/kafka/pull/12561#discussion_r962288239
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java:
##########
@@ -65,26 +67,32 @@ public class IncrementalCooperativeAssignor implements
ConnectAssignor {
private final int maxDelay;
private ConnectorsAndTasks previousAssignment;
private final ConnectorsAndTasks previousRevocation;
- private boolean canRevoke;
// visible for testing
+ boolean revokedInPrevious;
protected final Set<String> candidateWorkersForReassignment;
protected long scheduledRebalance;
protected int delay;
protected int previousGenerationId;
protected Set<String> previousMembers;
+ private final ExponentialBackoff consecutiveRevokingRebalancesBackoff;
+
+ private int numSuccessiveRevokingRebalances;
+
public IncrementalCooperativeAssignor(LogContext logContext, Time time,
int maxDelay) {
this.log = logContext.logger(IncrementalCooperativeAssignor.class);
this.time = time;
this.maxDelay = maxDelay;
this.previousAssignment = ConnectorsAndTasks.EMPTY;
this.previousRevocation = new ConnectorsAndTasks.Builder().build();
- this.canRevoke = true;
this.scheduledRebalance = 0;
+ this.revokedInPrevious = false;
this.candidateWorkersForReassignment = new LinkedHashSet<>();
this.delay = 0;
this.previousGenerationId = -1;
this.previousMembers = Collections.emptySet();
+ this.numSuccessiveRevokingRebalances = 0;
+ this.consecutiveRevokingRebalancesBackoff = new ExponentialBackoff(10,
20, maxDelay, 0.2);
Review Comment:
I added some logic to set delay to 0 when maxDelay is 0 in which case we
would always revoke. I bumped up the multiplier from 20-> 30 as well since
initial interval is reduced now.
--
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
