Skip to content

KAFKA-12495: IncrementalCooperativeAssignor#handleLostAssignments invokes logic for lost Assignments even when there are no Lost assignments #14000

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
vamossagar12 wants to merge 3 commits into from
Closed

KAFKA-12495: IncrementalCooperativeAssignor#handleLostAssignments invokes logic for lost Assignments even when there are no Lost assignments #14000

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5c7d6b8
Fixing comment with IncrementalCooperativeAssignor#handleLostAssignments
Jul 12, 2023
5c33d28
Adding changes to not invoke lost Assignments call when previous roun…
Jul 31, 2023
241cab2
Updated comments in tests
Jul 31, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 16 additions & 8 deletions ...ain/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,13 @@ ClusterAssignment performTaskAssignment(
previousRevocation.tasks().clear();
}

final long now = time.milliseconds();
// Before any operations, check if the rebalance delay has expired. In that case,
// we will reset the revoking rebalance state.
if (scheduledRebalance > 0 && now >= scheduledRebalance) {
resetPreviousRevocationsState();
}

// Derived set: The set of deleted connectors-and-tasks is a derived set from the set
// difference of previous - configured
ConnectorsAndTasks deleted = diff(previousAssignment, configured);
Expand Down Expand Up @@ -324,15 +331,14 @@ ClusterAssignment performTaskAssignment(
// have converged to a balanced load. We can reset the rebalance clock
log.debug("Previous round had revocations but this round didn't. Probably, the cluster has reached a " +
"balanced load. Resetting the exponential backoff clock");
revokedInPrevious = false;
numSuccessiveRevokingRebalances = 0;
resetPreviousRevocationsState();
} else {
// no-op
log.debug("No revocations in previous and current round.");
}
} else {
log.debug("Delayed rebalance is active. Delaying {}ms before revoking connectors and tasks: {}", delay, toRevoke);
revokedInPrevious = false;
resetPreviousRevocationsState();
}

// The complete set of connectors and tasks that should be newly-assigned during this round
Expand Down Expand Up @@ -442,8 +448,8 @@ private ConnectorsAndTasks duplicatedAssignments(Map<String, ConnectorsAndTasks>
protected void handleLostAssignments(ConnectorsAndTasks lostAssignments,
ConnectorsAndTasks.Builder lostAssignmentsToReassign,
List<WorkerLoad> completeWorkerAssignment) {
// There are no lost assignments and there have been no successive revoking rebalances
if (lostAssignments.isEmpty() && !revokedInPrevious) {

if (lostAssignments.isEmpty()) {
resetDelay();
return;
}
Expand Down Expand Up @@ -502,9 +508,6 @@ protected void handleLostAssignments(ConnectorsAndTasks lostAssignments,
lostAssignmentsToReassign.addTasks(lostAssignments.tasks());
}
resetDelay();
// Resetting the flag as now we can permit successive revoking rebalances.
// since we have gone through the full rebalance delay
revokedInPrevious = false;
} else {
candidateWorkersForReassignment
.addAll(candidateWorkersForReassignment(completeWorkerAssignment));
Expand Down Expand Up @@ -535,6 +538,11 @@ private void resetDelay() {
delay = 0;
}

private void resetPreviousRevocationsState() {
revokedInPrevious = false;
numSuccessiveRevokingRebalances = 0;
}

private Set<String> candidateWorkersForReassignment(List<WorkerLoad> completeWorkerAssignment) {
return completeWorkerAssignment.stream()
.filter(WorkerLoad::isEmpty)
Expand Down
24 changes: 16 additions & 8 deletions ...java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ public void testAssignmentsWhenWorkersJoinAfterRevocations() {
// in this round
addNewEmptyWorkers("worker3");
performStandardRebalance();
assertTrue(assignor.delay > 0);
assertEquals(40, assignor.delay); // First successive revoking rebalance.

@vamossagar12 vamossagar12 Jul 31, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added an elaborate check over here.

assertWorkers("worker1", "worker2", "worker3");
assertConnectorAllocations(0, 1, 2);
assertTaskAllocations(3, 3, 6);
Expand All @@ -161,30 +161,38 @@ public void testAssignmentsWhenWorkersJoinAfterRevocations() {
time.sleep(assignor.delay);
addNewEmptyWorkers("worker4");
performStandardRebalance();
assertEquals(0, assignor.delay); // No revocations in this round.
assertWorkers("worker1", "worker2", "worker3", "worker4");
assertConnectorAllocations(0, 0, 1, 1);
assertTaskAllocations(0, 3, 3, 3);

// Fifth assignment and a fifth worker joining after a revoking rebalance.
// We shouldn't revoke and set a delay > initial interval
// We shouldn't revoke and set a delay equal to initial interval
addNewEmptyWorkers("worker5");
performStandardRebalance();
assertTrue(assignor.delay > 40);
assertEquals(40, assignor.delay); // First successive revoking rebalance.

@vamossagar12 vamossagar12 Jul 31, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IMO this happens because we weren't unsetting the counter when delay goes equal to 0. This is the first successive revoking rebalance and the delay should be 40. I say this because this case is similar to worker2 -> worker3 joining above where the delay was 40. It should be similar here as well.

assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5");
assertConnectorAllocations(0, 0, 1, 1, 1);
assertTaskAllocations(1, 2, 3, 3, 3);

// Sixth assignment with sixth worker joining after the expiry.
// Should revoke
time.sleep(assignor.delay);

@vamossagar12 vamossagar12 Jul 31, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I removed this to prove that the changes still work correctly and the delay gets extended further.

// Sixth assignment with sixth worker joining within rebalance interval
// There should not be any revocations and delay should be extended further.
addNewEmptyWorkers("worker6");
performStandardRebalance();
assertDelay(0);
assertTrue(assignor.delay > 40); // Second successive revoking rebalance.
assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5", "worker6");
assertConnectorAllocations(0, 0, 0, 1, 1, 1);
assertTaskAllocations(0, 1, 2, 3, 3, 3);

// Follow up rebalance after delay expires.
time.sleep(assignor.delay);
performStandardRebalance();
assertEquals(0, assignor.delay); // Delay reset
assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5", "worker6");
assertConnectorAllocations(0, 0, 0, 1, 1, 1);
assertTaskAllocations(0, 1, 2, 2, 2, 2);

// Follow up rebalance since there were revocations
// Final rebalance because of revocations in the previous round
performStandardRebalance();
assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5", "worker6");
assertConnectorAllocations(0, 0, 0, 1, 1, 1);
Expand Down