lucasbru commented on code in PR #23482:
URL: https://github.com/apache/kafka/pull/23482#discussion_r4062202152
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -414,9 +449,12 @@ private static Member findPrevMemberWithLeastLoad(
private static boolean hasUnfulfilledActiveTaskQuota(
final LocalState localState,
final ProcessState process,
- final Member member
+ final Member member,
Review Comment:
nit: This quota check only works because assignActive(stateful) always runs
before assignActive(stateless)/assignStandby, so the member's task count still
equals its stateful count at this point. Nothing enforces that ordering though
- worth a comment calling out the invariant, so a future reorder doesn't
silently corrupt the quota check?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -68,56 +77,54 @@ public String toString() {
public GroupAssignment assign(final GroupSpec groupSpec, final
TopologyDescriber topologyDescriber) throws TaskAssignorException {
return doAssign(
initialize(groupSpec, topologyDescriber),
- groupSpec,
- topologyDescriber
+ groupSpec
);
}
private static GroupAssignment doAssign(
final LocalState localState,
- final GroupSpec groupSpec,
- final TopologyDescriber topologyDescriber
+ final GroupSpec groupSpec
) {
- final LinkedList<TaskId> activeTasks = taskIds(topologyDescriber,
true);
- assignActive(localState, activeTasks);
+ // Stateful and stateless active tasks are balanced independently: the
stateful ones are placed first, then
+ // the stateless ones fill up the remaining active capacity.
+ assignActive(localState, new
LinkedList<>(localState.statefulActiveTaskIds), true);
+ assignActive(localState, new
LinkedList<>(localState.statelessActiveTaskIds), false);
if (localState.numStandbyReplicas > 0) {
- final LinkedList<TaskId> statefulTasks =
taskIds(topologyDescriber, false);
- assignStandby(localState, statefulTasks);
+ assignStandby(localState, new
LinkedList<>(localState.statefulActiveTaskIds));
Review Comment:
Do we need to copy all three task lists here? assignActive consumes its
input, but assignStandby only sorts and iterates its list. Could we preserve
statefulActiveTaskIds for standby assignment and pass the stateless list
directly, avoiding the extra LinkedList copy?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
