This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 362a0371c6e KAFKA-20198: Fix StickyTaskAssignor capacity calculation
to use proportional instance limit (#22006)
362a0371c6e is described below
commit 362a0371c6e34e12830b666045f4ff0d283e3fe4
Author: Mingi Cho <[email protected]>
AuthorDate: Fri Jul 24 06:38:39 2026 +0900
KAFKA-20198: Fix StickyTaskAssignor capacity calculation to use
proportional instance limit (#22006)
[KAFKA-20198](https://issues.apache.org/jira/browse/KAFKA-20198)
Saw the discussion on Confluent Slack about `StickyTaskAssignor` taking
many rebalance rounds to converge with the classic group protocol when
scaling up. Dug into it and found the root cause in
`hasRoomForActiveTask`.
The per-instance limit is computed as `capacity * (taskCount /
totalCapacity)`, but the integer division floors — so with 450 tasks and
20 threads, each instance gets a limit of `10 * 22 = 220` instead of the
fair share 225. The 10-task gap pushes overflow into
`findBestClientForTask`, which picks differently each round due to
HashMap iteration order, creating a feedback loop that prevents
convergence.
Fixed by computing the instance limit directly as a proportional share:
`(taskCount * capacity + totalCapacity - 1) / totalCapacity`, giving
`ceil(450 * 10 / 20) = 225`. Same approach as
`AbstractStickyAssignor.maxQuota`.
Simulated with the actual `StickyTaskAssignor` across cooperative
rebalance rounds:
| Scenario | Before | After | |----------|--------|-------| | 450p/10t |
5 rounds | 2 rounds | | 100p/4t | 4 rounds | 2 rounds | | 200p/8t | 6
rounds | 2 rounds | | 1000p/16t | 9 rounds | 5 rounds | | 500p/12t | 9
rounds | 3 rounds | | Even cases | 2 rounds | 2 rounds |
All existing tests pass.
Reviewers: gabriellafu (github:gabriellefu), Hari Mohan (github:harimm),
Bill Bejeck <[email protected]>
---
.../assignment/assignors/StickyTaskAssignor.java | 10 +--
.../assignment/CustomStickyTaskAssignorTest.java | 77 ++++++++++++++++++++++
2 files changed, 82 insertions(+), 5 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/assignment/assignors/StickyTaskAssignor.java
b/streams/src/main/java/org/apache/kafka/streams/processor/assignment/assignors/StickyTaskAssignor.java
index ec16c0dde3d..0e7c80a30b9 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/assignment/assignors/StickyTaskAssignor.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/assignment/assignors/StickyTaskAssignor.java
@@ -148,7 +148,6 @@ public class StickyTaskAssignor implements TaskAssignor {
final int totalCapacity = computeTotalProcessingThreads(clients);
final Set<TaskId> allTaskIds = applicationState.allTasks().keySet();
final int taskCount = allTaskIds.size();
- final int activeTasksPerThread = taskCount / totalCapacity;
final Set<TaskId> unassigned = new HashSet<>(allTaskIds);
// first try and re-assign existing active tasks to clients that
previously had
@@ -156,7 +155,7 @@ public class StickyTaskAssignor implements TaskAssignor {
for (final TaskId taskId :
assignmentState.previousActiveAssignment.keySet()) {
final ProcessId previousClientForTask =
assignmentState.previousActiveAssignment.get(taskId);
if (allTaskIds.contains(taskId)) {
- if (mustPreserveActiveTaskAssignment ||
assignmentState.hasRoomForActiveTask(previousClientForTask,
activeTasksPerThread)) {
+ if (mustPreserveActiveTaskAssignment ||
assignmentState.hasRoomForActiveTask(previousClientForTask, taskCount,
totalCapacity)) {
assignmentState.finalizeAssignment(taskId,
previousClientForTask, AssignedTask.Type.ACTIVE);
unassigned.remove(taskId);
}
@@ -169,7 +168,7 @@ public class StickyTaskAssignor implements TaskAssignor {
final TaskId taskId = iterator.next();
final Set<ProcessId> previousClientsForStandbyTask =
assignmentState.previousStandbyAssignment.getOrDefault(taskId, new HashSet<>());
for (final ProcessId client: previousClientsForStandbyTask) {
- if (assignmentState.hasRoomForActiveTask(client,
activeTasksPerThread)) {
+ if (assignmentState.hasRoomForActiveTask(client, taskCount,
totalCapacity)) {
assignmentState.finalizeAssignment(taskId, client,
AssignedTask.Type.ACTIVE);
iterator.remove();
break;
@@ -297,14 +296,15 @@ public class StickyTaskAssignor implements TaskAssignor {
this.newAssignments = optimizedAssignments;
}
- private boolean hasRoomForActiveTask(final ProcessId processId, final
int activeTasksPerThread) {
+ private boolean hasRoomForActiveTask(final ProcessId processId, final
int taskCount, final int totalCapacity) {
final int capacity = clients.get(processId).numProcessingThreads();
final int newActiveTaskCount =
newAssignments.computeIfAbsent(processId, k ->
KafkaStreamsAssignment.of(processId, new HashSet<>()))
.tasks().values()
.stream().filter(assignedTask -> assignedTask.type() ==
AssignedTask.Type.ACTIVE)
.collect(Collectors.toSet())
.size();
- return newActiveTaskCount < capacity * activeTasksPerThread;
+ final int instanceLimit = (taskCount * capacity + totalCapacity -
1) / totalCapacity;
+ return newActiveTaskCount < instanceLimit;
}
private ProcessId findBestClientForTask(final TaskId taskId, final
Set<ProcessId> clientsWithin) {
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/CustomStickyTaskAssignorTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/CustomStickyTaskAssignorTest.java
index fa92a55dcae..a9a71d16f59 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/CustomStickyTaskAssignorTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/CustomStickyTaskAssignorTest.java
@@ -39,6 +39,8 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -78,6 +80,7 @@ import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.fail;
public class CustomStickyTaskAssignorTest {
@@ -913,4 +916,78 @@ public class CustomStickyTaskAssignorTest {
assertThat(topicGroupIds, equalTo(asList(1, 2)));
}
}
+
+ // --- KAFKA-20198 regression tests ---
+
+ @Test
+ public void shouldRetainFairShareOfPreviousTasksWhenScalingUp() {
+ final int numTasks = 450;
+ final int threadsPerInstance = 10;
+ final Map<TaskId, TaskInfo> tasks = buildStatelessTasks(numTasks);
+ final Set<TaskId> firstHalf = buildTaskIdRange(0, numTasks / 2);
+
+ final Map<ProcessId, KafkaStreamsState> streamStates = mkMap(
+ mkStreamState(1, threadsPerInstance, Optional.empty(), firstHalf,
Set.of()),
+ mkStreamState(2, threadsPerInstance, Optional.empty())
+ );
+
+ final Map<ProcessId, KafkaStreamsAssignment> assignments =
+ assign(streamStates, tasks,
StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_NONE);
+
+ final Set<TaskId> instance1Tasks = activeTasks(assignments, 1);
+ final int retained = (int)
firstHalf.stream().filter(instance1Tasks::contains).count();
+
+ assertThat("Instance should retain all of its fair share tasks, but
only retained " + retained + " of " + firstHalf.size(),
+ retained, equalTo(firstHalf.size()));
+ }
+
+ @Test
+ public void shouldConvergeWithinTwoRoundsWhenScalingUp() {
+ // Reporter's scenario: 450 tasks with 10+10=20 threads.
+ // 450/20 = 22.5, floor gives limit 220 per instance (10 task
overflow),
+ // causing repeated reassignment across rounds.
+ final int numTasks = 450;
+ final int maxRounds = 2;
+ final Map<TaskId, TaskInfo> tasks = buildStatelessTasks(numTasks);
+
+ Set<TaskId> instance1Prev = buildTaskIdRange(0, numTasks);
+ Set<TaskId> instance2Prev = Set.of();
+
+ for (int round = 1; round <= maxRounds; round++) {
+ final Map<ProcessId, KafkaStreamsState> streamStates = mkMap(
+ mkStreamState(1, 10, Optional.empty(), instance1Prev,
Set.of()),
+ mkStreamState(2, 10, Optional.empty(), instance2Prev, Set.of())
+ );
+
+ final Map<ProcessId, KafkaStreamsAssignment> assignments =
+ assign(streamStates, tasks,
StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_NONE);
+
+ final Set<TaskId> newInstance1 = activeTasks(assignments, 1);
+ final Set<TaskId> newInstance2 = activeTasks(assignments, 2);
+
+ if (newInstance1.equals(instance1Prev) &&
newInstance2.equals(instance2Prev)) {
+ return; // converged
+ }
+ instance1Prev = newInstance1;
+ instance2Prev = newInstance2;
+ }
+ fail("Assignment did not converge within " + maxRounds + " rounds");
+ }
+
+ private static Map<TaskId, TaskInfo> buildStatelessTasks(final int count) {
+ final Map<TaskId, TaskInfo> tasks = new HashMap<>();
+ for (int i = 0; i < count; i++) {
+ final TaskId taskId = new TaskId(0, i);
+ tasks.put(taskId, mkTaskInfo(taskId, false).getValue());
+ }
+ return tasks;
+ }
+
+ private static Set<TaskId> buildTaskIdRange(final int from, final int to) {
+ final Set<TaskId> set = new HashSet<>();
+ for (int i = from; i < to; i++) {
+ set.add(new TaskId(0, i));
+ }
+ return set;
+ }
}
