gabriellefu commented on code in PR #23482:
URL: https://github.com/apache/kafka/pull/23482#discussion_r4065187918
##########
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:
I have updated my pr, thanks
--
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]
