You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2025/09/02 12:30:11 UTC
[PR] KAFKA-19661 [4/N]: Use below-quota as a condition for standby task assignment [kafka]
lucasbru opened a new pull request, #20458:
URL: https://github.com/apache/kafka/pull/20458
In the original algorithm, standby tasks are assigned to a
process that previously owned the task only if it is
“load-balanced”, meaning the process has fewer tasks that
members, or it is the least loaded process. This is strong
requirement, and will cause standby tasks to often not get
assigned to process that previously owned it. Furthermore,
the condition “is the least loaded process” is hard to
evaluate efficiently in this context.
We propose to instead use the same “below-quota” condition
as in active task assignment.
We compute a quota for active and standby tasks, by definiing
numOfTasks = numberOfActiveTasks+numberOfStandbyTasks and
defining the quota as numOfTasks/numberOfMembers rounded up.
Whenever a member becomes “full” (its assigned number of tasks
is equal to numOfTasks) we deduct its tasks from numOfTasks and
decrement numberOfMembers and recompute the quota, which means
that the quota may be reduced by one during the assignment
process, to avoid uneven assignments.
A standby task can be assigned to a process that previously
owned it, whenever the process has fewer than
numOfMembersOfProcess*quota.
This condition will, again, prioritize standby stickyness,
and can be evaluated in constant time.
In our worst-performing benchmark, this improves the runtime
by 2.5x on top of the previous optimizations, but 5x on the
more important incremental assignment case.
# This is a stacked PR. Only review the last commit
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19661 [5/N]: Use below-quota as a condition for standby task assignment [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20458:
URL: https://github.com/apache/kafka/pull/20458#discussion_r2319150858
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -53,24 +55,19 @@ public GroupAssignment assign(final GroupSpec groupSpec, final TopologyDescriber
}
private GroupAssignment doAssign(final GroupSpec groupSpec, final TopologyDescriber topologyDescriber) {
- //active
- final Set<TaskId> activeTasks = taskIds(topologyDescriber, true);
+ final Deque<TaskId> activeTasks = taskIds(topologyDescriber, true);
assignActive(activeTasks);
- //standby
- final int numStandbyReplicas =
- groupSpec.assignmentConfigs().isEmpty() ? 0
- : Integer.parseInt(groupSpec.assignmentConfigs().get("num.standby.replicas"));
- if (numStandbyReplicas > 0) {
- final Set<TaskId> statefulTasks = taskIds(topologyDescriber, false);
- assignStandby(statefulTasks, numStandbyReplicas);
+ if (localState.numStandbyReplicas > 0) {
+ final Deque<TaskId> statefulTasks = taskIds(topologyDescriber, false);
+ assignStandby(statefulTasks);
}
return buildGroupAssignment(groupSpec.members().keySet());
}
- private Set<TaskId> taskIds(final TopologyDescriber topologyDescriber, final boolean isActive) {
- final Set<TaskId> ret = new HashSet<>();
+ private Deque<TaskId> taskIds(final TopologyDescriber topologyDescriber, final boolean isActive) {
Review Comment:
We want to assign standby tasks in reverse, so I'm using a Deque here, which provides a reverseIterator.
The reason why we want to traverse standby tasks in reverse is the example that I added in the unit tests of both LegacyStickTaskAssignor and the new StickyTaskAssignor.
Assume we have
Node 1: Active task 0,1, Standby task 2,3
Node 2: Active task 2,3, Standby task 0,1
Node 3: - (new)
Then we don't want to assign active tasks and standby tasks in the same order.
Suppose we try to assign active tasks in increasing order, we will get:
Node 1: Active task 0,1
Node 2: Active task 2
Node 3: Active task 3
Since task 3 is the last task we will assign, and at that point, the quota for active tasks is 1, so it can only be assigned to Node 3.
Suppose now we assign standby tasks in the same order, we will get this:
Node 1: Active task 0,1, Standby task 2, 3
Node 2: Active task 2, Standby task 0, 1
Node 3: Active task 3
The reason is that we first assign tasks 0,1,2, which all can be assigned to the previous member that owned it. Finally, we want to assign standby task 3, but it cannot be assigned to Node 3, so we have to assign it to Node 1 or Node 2. Using reverse order means, when I have new nodes, they will get the numerically last few active tasks, and the numerically first standby tasks, which should avoid this problem.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19661 [4/N]: Use below-quota as a condition for standby task assignment [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20458:
URL: https://github.com/apache/kafka/pull/20458#discussion_r2315959967
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -185,11 +195,13 @@ private void assignActive(final Set<TaskId> activeTasks) {
final TaskId task = it.next();
final ArrayList<Member> prevMembers = localState.standbyTaskToPrevMember.get(task);
final Member prevMember = findPrevMemberWithLeastLoad(prevMembers, null);
- if (prevMember != null && hasUnfulfilledQuota(prevMember)) {
+ if (prevMember != null) {
final ProcessState processState = localState.processIdToState.get(prevMember.processId);
- processState.addTask(prevMember.memberId, task, true);
- maybeUpdateTasksPerMember(processState.activeTaskCount());
- it.remove();
+ if (hasUnfulfilledActiveTaskQuota(processState, prevMember)) {
Review Comment:
as a minor optimization, we passed `processState` into this function (to avoid fetching it again inside), so the if construct changed here slightly.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -53,24 +55,19 @@ public GroupAssignment assign(final GroupSpec groupSpec, final TopologyDescriber
}
private GroupAssignment doAssign(final GroupSpec groupSpec, final TopologyDescriber topologyDescriber) {
- //active
- final Set<TaskId> activeTasks = taskIds(topologyDescriber, true);
+ final Deque<TaskId> activeTasks = taskIds(topologyDescriber, true);
assignActive(activeTasks);
- //standby
- final int numStandbyReplicas =
- groupSpec.assignmentConfigs().isEmpty() ? 0
- : Integer.parseInt(groupSpec.assignmentConfigs().get("num.standby.replicas"));
- if (numStandbyReplicas > 0) {
- final Set<TaskId> statefulTasks = taskIds(topologyDescriber, false);
- assignStandby(statefulTasks, numStandbyReplicas);
+ if (localState.numStandbyReplicas > 0) {
+ final Deque<TaskId> statefulTasks = taskIds(topologyDescriber, false);
+ assignStandby(statefulTasks);
}
return buildGroupAssignment(groupSpec.members().keySet());
}
- private Set<TaskId> taskIds(final TopologyDescriber topologyDescriber, final boolean isActive) {
- final Set<TaskId> ret = new HashSet<>();
+ private Deque<TaskId> taskIds(final TopologyDescriber topologyDescriber, final boolean isActive) {
Review Comment:
We want to assign standby tasks in reverse, so I'm using a Deque here, which provides a reverseIterator.
The reason why we want to traverse standby tasks in reverse is the example that I added in the unit tests of both LegacyStickTaskAssignor and the new StickyTaskAssignor.
Assume we have
Node 1: Active task 0,1, Standby task 2,3
Node 2: Active task 2,3, Standby task 0,1
Node 3: - (new)
Then we don't want to assign active tasks and standby tasks in the same order.
Suppose we try to assign active tasks in increasing order, we will get:
Node 1: Active task 0,1
Node 2: Active task 2
Node 3: Active task 3
Since task 3 is the last task we will assign, and at that point, the quota for active tasks is 1, so it can only be assigned to Node 3.
Suppose now we assign standby tasks in the same order, we will get this:
Node 1: Active task 0,1, Standby task 2, 3
Node 2: Active task 2, Standby task 0, 1
Node 3: Active task 3
The reason is that we first assign tasks 0,1,2, which all can be assigned to the previous member that owned it. Finally, we want to assign standby task 3, but it cannot be assigned to Node 3, so we have to assign it to Node 1 or Node 2. Using reverse order means, when I have new nodes, they will get the numerically last few active tasks, and the numerically first standby tasks, which should avoid this problem.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -84,13 +81,24 @@ private Set<TaskId> taskIds(final TopologyDescriber topologyDescriber, final boo
private void initialize(final GroupSpec groupSpec, final TopologyDescriber topologyDescriber) {
localState = new LocalState();
- localState.allTasks = 0;
Review Comment:
`allTasks` was previously just referring to active tasks.
I renamed it to `totalActiveTasks`, and also added `totalTasks` that counts the number of tasks including standby tasks.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -58,12 +58,9 @@ private GroupAssignment doAssign(final GroupSpec groupSpec, final TopologyDescri
assignActive(activeTasks);
//standby
- final int numStandbyReplicas =
Review Comment:
This variable just moved to "localState" to avoid passing it around.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19661 [5/N]: Use below-quota as a condition for standby task assignment [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #20458:
URL: https://github.com/apache/kafka/pull/20458#discussion_r2332567380
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -84,13 +80,24 @@ private LinkedList<TaskId> taskIds(final TopologyDescriber topologyDescriber, fi
private void initialize(final GroupSpec groupSpec, final TopologyDescriber topologyDescriber) {
localState = new LocalState();
- localState.allTasks = 0;
+ localState.numStandbyReplicas =
+ groupSpec.assignmentConfigs().isEmpty() ? 0
+ : Integer.parseInt(groupSpec.assignmentConfigs().get("num.standby.replicas"));
+
+ // Helpers for computing active tasks per member, and tasks per member
+ localState.totalActiveTasks = 0;
+ localState.totalTasks = 0;
for (final String subtopology : topologyDescriber.subtopologies()) {
final int numberOfPartitions = topologyDescriber.maxNumInputPartitions(subtopology);
- localState.allTasks += numberOfPartitions;
+ localState.totalTasks += numberOfPartitions;
+ localState.totalActiveTasks += numberOfPartitions;
+ if (topologyDescriber.isStateful(subtopology))
+ localState.totalTasks += numberOfPartitions * localState.numStandbyReplicas;
Review Comment:
Missing braces around the if statement body. According to Java coding standards, single-line if statements should include braces for clarity and to prevent errors during future modifications.
```suggestion
if (topologyDescriber.isStateful(subtopology)) {
localState.totalTasks += numberOfPartitions * localState.numStandbyReplicas;
}
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -298,43 +317,49 @@ private String memberWithLeastLoad(final ProcessState processWithLeastLoad) {
return memberWithLeastLoad.orElse(null);
}
- private boolean hasUnfulfilledQuota(final Member member) {
- return localState.processIdToState.get(member.processId).memberToTaskCounts().get(member.memberId) < localState.tasksPerMember;
+ private boolean hasUnfulfilledActiveTaskQuota(final ProcessState process, final Member member) {
+ return process.memberToTaskCounts().get(member.memberId) < localState.activeTasksPerMember;
}
- private void assignStandby(final LinkedList<TaskId> standbyTasks, int numStandbyReplicas) {
- final ArrayList<StandbyToAssign> toLeastLoaded = new ArrayList<>(standbyTasks.size() * numStandbyReplicas);
+ private boolean hasUnfulfilledTaskQuota(final ProcessState process, final Member member) {
+ return process.memberToTaskCounts().get(member.memberId) < localState.tasksPerMember;
+ }
+ private void assignStandby(final LinkedList<TaskId> standbyTasks) {
+ final ArrayList<StandbyToAssign> toLeastLoaded = new ArrayList<>(standbyTasks.size() * localState.numStandbyReplicas);
+
Review Comment:
Trailing whitespace should be removed from line 330.
```suggestion
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19661 [5/N]: Use below-quota as a condition for standby task assignment [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20458:
URL: https://github.com/apache/kafka/pull/20458#discussion_r2322407061
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -166,17 +174,22 @@ private Map<String, Set<Integer>> toCompactedTaskIds(final Set<TaskId> taskIds)
return ret;
}
- private void assignActive(final Set<TaskId> activeTasks) {
+ private void assignActive(final LinkedList<TaskId> activeTasks) {
+
+ // Assuming our current assignment is range-based, we want to sort by partition first
+ activeTasks.sort(Comparator.comparing(TaskId::partition).thenComparing(TaskId::subtopologyId));
Review Comment:
I added the sorting here. The old assignor did not do the sorting explicitly, but randomly ran into the "good case".
The point is this:
Normally, we want to assign an active task like a range-assignor, when we have two subtopologies with two partitions and two clients, we will assign
Client1: 0_0, 1_0
Client2: 0_1, 1_1
The reason being, heuristically, if we'd have the assignment
Client1: 0_0, 0_1
Client2: 1_0, 1_1
and the first subtopology has large state and the second subtopology has small state, then one client gets most of the state.
The sorting here helps to also achieve this kind of range assignment when scaling up. Assume we have now all tasks assigned to the first member:
Client1: 0_0, 0_1, 1_0, 1_1
Client2: -
Now, we will first assign the previous tasks, we want to start with all 0 partitions, before doing all 1 partitions, until Client1 fills up:
Client1: 0_0, 1_0
Client2:
Then filling upp client2 the usual way.
Client1: 0_0, 1_0
Client2: 1_0, 1_1
This is corner case, but seems like a useful improvement.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19661 [5/N]: Use below-quota as a condition for standby task assignment [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #20458:
URL: https://github.com/apache/kafka/pull/20458
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19661 [4/N]: Use below-quota as a condition for standby task assignment [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #20458:
URL: https://github.com/apache/kafka/pull/20458#discussion_r2318899271
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -103,19 +113,19 @@ private void initialize(final GroupSpec groupSpec, final TopologyDescriber topol
localState.processIdToState.get(processId).addMember(memberId);
// prev active tasks
- for (Map.Entry<String, Set<Integer>> entry : memberSpec.activeTasks().entrySet()) {
- Set<Integer> partitionNoSet = entry.getValue();
- for (int partitionNo : partitionNoSet) {
+ for (final Map.Entry<String, Set<Integer>> entry : memberSpec.activeTasks().entrySet()) {
+ final Set<Integer> partitionNoSet = entry.getValue();
+ for (final int partitionNo : partitionNoSet) {
localState.activeTaskToPrevMember.put(new TaskId(entry.getKey(), partitionNo), member);
}
}
// prev standby tasks
- for (Map.Entry<String, Set<Integer>> entry : memberSpec.standbyTasks().entrySet()) {
- Set<Integer> partitionNoSet = entry.getValue();
- for (int partitionNo : partitionNoSet) {
- TaskId taskId = new TaskId(entry.getKey(), partitionNo);
- localState.standbyTaskToPrevMember.putIfAbsent(taskId, new HashSet<>());
+ for (final Map.Entry<String, Set<Integer>> entry : memberSpec.standbyTasks().entrySet()) {
+ final Set<Integer> partitionNoSet = entry.getValue();
+ for (final int partitionNo : partitionNoSet) {
+ final TaskId taskId = new TaskId(entry.getKey(), partitionNo);
+ localState.standbyTaskToPrevMember.putIfAbsent(taskId, new ArrayList<>());
Review Comment:
The change from `Set<Member>` to `ArrayList<Member>` for `standbyTaskToPrevMember` allows duplicate members to be added. This could lead to incorrect behavior when multiple standby replicas point to the same member for a task.
```suggestion
localState.standbyTaskToPrevMember.putIfAbsent(taskId, new HashSet<>());
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -156,143 +166,207 @@ private GroupAssignment buildGroupAssignment(final Set<String> members) {
}
private Map<String, Set<Integer>> toCompactedTaskIds(final Set<TaskId> taskIds) {
- Map<String, Set<Integer>> ret = new HashMap<>();
- for (TaskId taskId : taskIds) {
+ final Map<String, Set<Integer>> ret = new HashMap<>();
+ for (final TaskId taskId : taskIds) {
ret.putIfAbsent(taskId.subtopologyId(), new HashSet<>());
ret.get(taskId.subtopologyId()).add(taskId.partition());
}
return ret;
}
- private void assignActive(final Set<TaskId> activeTasks) {
+ private void assignActive(final Deque<TaskId> activeTasks) {
// 1. re-assigning existing active tasks to clients that previously had the same active tasks
- for (Iterator<TaskId> it = activeTasks.iterator(); it.hasNext();) {
+ for (final Iterator<TaskId> it = activeTasks.iterator(); it.hasNext();) {
final TaskId task = it.next();
final Member prevMember = localState.activeTaskToPrevMember.get(task);
- if (prevMember != null && hasUnfulfilledQuota(prevMember)) {
- localState.processIdToState.get(prevMember.processId).addTask(prevMember.memberId, task, true);
- updateHelpers(prevMember, true);
- it.remove();
+ if (prevMember != null) {
+ final ProcessState processState = localState.processIdToState.get(prevMember.processId);
+ if (hasUnfulfilledActiveTaskQuota(processState, prevMember)) {
+ processState.addTask(prevMember.memberId, task, true);
+ maybeUpdateActiveTasksPerMember(processState.memberToTaskCounts().get(prevMember.memberId));
+ it.remove();
+ }
}
}
// 2. re-assigning tasks to clients that previously have seen the same task (as standby task)
- for (Iterator<TaskId> it = activeTasks.iterator(); it.hasNext();) {
+ for (final Iterator<TaskId> it = activeTasks.iterator(); it.hasNext();) {
final TaskId task = it.next();
- final Set<Member> prevMembers = localState.standbyTaskToPrevMember.get(task);
- final Member prevMember = findMemberWithLeastLoad(prevMembers, task, true);
- if (prevMember != null && hasUnfulfilledQuota(prevMember)) {
- localState.processIdToState.get(prevMember.processId).addTask(prevMember.memberId, task, true);
- updateHelpers(prevMember, true);
- it.remove();
+ final ArrayList<Member> prevMembers = localState.standbyTaskToPrevMember.get(task);
+ final Member prevMember = findPrevMemberWithLeastLoad(prevMembers, null);
+ if (prevMember != null) {
+ final ProcessState processState = localState.processIdToState.get(prevMember.processId);
+ if (hasUnfulfilledActiveTaskQuota(processState, prevMember)) {
+ processState.addTask(prevMember.memberId, task, true);
+ maybeUpdateActiveTasksPerMember(processState.memberToTaskCounts().get(prevMember.memberId));
+ it.remove();
+ }
}
}
// 3. assign any remaining unassigned tasks
- for (Iterator<TaskId> it = activeTasks.iterator(); it.hasNext();) {
+ final PriorityQueue<ProcessState> processByLoad = new PriorityQueue<>(Comparator.comparingDouble(ProcessState::load));
+ processByLoad.addAll(localState.processIdToState.values());
+ for (final Iterator<TaskId> it = activeTasks.iterator(); it.hasNext();) {
final TaskId task = it.next();
- final Set<Member> allMembers = localState.processIdToState.entrySet().stream().flatMap(entry -> entry.getValue().memberToTaskCounts().keySet().stream()
- .map(memberId -> new Member(entry.getKey(), memberId))).collect(Collectors.toSet());
- final Member member = findMemberWithLeastLoad(allMembers, task, false);
+ final ProcessState processWithLeastLoad = processByLoad.poll();
+ if (processWithLeastLoad == null) {
+ throw new TaskAssignorException("No process available to assign active task {}." + task);
+ }
+ final String member = memberWithLeastLoad(processWithLeastLoad);
if (member == null) {
- log.error("Unable to assign active task {} to any member.", task);
throw new TaskAssignorException("No member available to assign active task {}." + task);
}
- localState.processIdToState.get(member.processId).addTask(member.memberId, task, true);
+ processWithLeastLoad.addTask(member, task, true);
it.remove();
- updateHelpers(member, true);
+ maybeUpdateActiveTasksPerMember(processWithLeastLoad.memberToTaskCounts().get(member));
+ processByLoad.add(processWithLeastLoad); // Add it back to the queue after updating its state
+ }
+ }
+
+ private void maybeUpdateActiveTasksPerMember(final int activeTasksNo) {
+ if (activeTasksNo == localState.activeTasksPerMember) {
+ localState.totalMembersWithActiveTaskCapacity--;
+ localState.totalActiveTasks -= activeTasksNo;
+ localState.activeTasksPerMember = computeTasksPerMember(localState.totalActiveTasks, localState.totalMembersWithActiveTaskCapacity);
+ }
+ }
+ private void maybeUpdateTasksPerMember(final int taskNo) {
+ if (taskNo == localState.tasksPerMember) {
+ localState.totalMembersWithTaskCapacity--;
+ localState.totalTasks -= taskNo;
+ localState.tasksPerMember = computeTasksPerMember(localState.totalTasks, localState.totalMembersWithTaskCapacity);
}
}
- private void maybeUpdateTasksPerMember(final int activeTasksNo) {
- if (activeTasksNo == localState.tasksPerMember) {
- localState.totalCapacity--;
- localState.allTasks -= activeTasksNo;
- localState.tasksPerMember = computeTasksPerMember(localState.allTasks, localState.totalCapacity);
+ private boolean assignStandbyToMemberWithLeastLoad(PriorityQueue<ProcessState> queue, TaskId taskId) {
+ final ProcessState processWithLeastLoad = queue.poll();
+ if (processWithLeastLoad == null) {
+ return false;
+ }
+ boolean found = false;
+ if (!processWithLeastLoad.hasTask(taskId)) {
+ final String memberId = memberWithLeastLoad(processWithLeastLoad);
+ if (memberId != null) {
+ processWithLeastLoad.addTask(memberId, taskId, false);
+ found = true;
+ }
+ } else if (!queue.isEmpty()) {
+ found = assignStandbyToMemberWithLeastLoad(queue, taskId);
}
+ queue.add(processWithLeastLoad); // Add it back to the queue after updating its state
+ return found;
}
- private Member findMemberWithLeastLoad(final Set<Member> members, TaskId taskId, final boolean returnSameMember) {
+ /**
+ * Finds the previous member with the least load for a given task.
+ *
+ * @param members The list of previous members owning the task.
+ * @param taskId The taskId, to check if the previous member already has the task. Can be null, if we assign it
+ * for the first time (e.g., during active task assignment).
+ *
+ * @return Previous member with the least load that deoes not have the task, or null if no such member exists.
Review Comment:
There's a typo in the documentation comment: 'deoes' should be 'does'.
```suggestion
* @return Previous member with the least load that does not have the task, or null if no such member exists.
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19661 [4/N]: Use below-quota as a condition for standby task assignment [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20458:
URL: https://github.com/apache/kafka/pull/20458#discussion_r2322407061
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -166,17 +174,22 @@ private Map<String, Set<Integer>> toCompactedTaskIds(final Set<TaskId> taskIds)
return ret;
}
- private void assignActive(final Set<TaskId> activeTasks) {
+ private void assignActive(final LinkedList<TaskId> activeTasks) {
+
+ // Assuming our current assignment is range-based, we want to sort by partition first
+ activeTasks.sort(Comparator.comparing(TaskId::partition).thenComparing(TaskId::subtopologyId));
Review Comment:
I added the sorting here. The old assignor did not do the sorting explicitly, but randomly ran into the "good case".
The point is this:
Normally, we want to assign an active task like a range-assignor, when we have two subtopologies with two partitions and two clients, we will assign
Client1: 0_0, 1_0
Client2: 0_1, 1_1
The reason being, heuristically, if we'd have the assignment
Client1: 0_0, 0_1
Client2: 1_0, 1_1
and the first subtopology has large state and the second subtopology has small state, then one client gets most of the state.
The sorting here helps to also achieve this kind of range assignment when scaling up. Assume we have now all tasks assigned to the first member:
Client1: 0_0, 0_1, 1_0, 1_1
Client2: -
Now, we will first assign the previous tasks, we want to start with all 0 partitions, before doing all 1 partitions, until Client1 fills up:
Client1: 0_0, 1_0
Client2:
Then filling upp client2 the usual way.
Client1: 0_0, 1_0
Client2: 1_0, 1_1
This is corner case, but seems like a useful improvement.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org