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 all 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
25 changes: 17 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 @@ -142,6 +142,7 @@ public void testAssignmentsWhenWorkersJoinAfterRevocations() {
// We should revoke.
addNewEmptyWorkers("worker2");
performStandardRebalance();
assertEquals(0, assignor.delay); // First revoking rebalance
assertWorkers("worker1", "worker2");
assertConnectorAllocations(0, 2);
assertTaskAllocations(0, 6);
Expand All @@ -151,7 +152,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 +162,38 @@ public void testAssignmentsWhenWorkersJoinAfterRevocations() {
time.sleep(assignor.delay);
addNewEmptyWorkers("worker4");
performStandardRebalance();
assertEquals(0, assignor.delay); // First revoking rebalance after delay
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 after delay
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 after delay
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