lucasbru commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4024825537
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -129,6 +132,58 @@ private static boolean isRestoring(final MemberTaskOffsets
memberTaskOffsets, fi
return offsetOf(memberTaskOffsets.taskOffsets(), task) != null;
}
+ /**
+ * Indexes how loaded each process is, for the order in which the budget
pass funds warm-up tasks. The load of a
+ * process is its stateful task count over the number of members it runs
-- the same shape as the task assignor's
+ * own {@code ProcessState.load()}, so that both layers rank processes
comparably.
+ *
+ * <p><b>Only stateful tasks are counted</b>, which is narrower than what
the assignor measures. Standby and
+ * warm-up tasks exist only for stateful tasks anyway, so in practice this
comes down to leaving stateless active
+ * tasks out, for two reasons. Where the assignor spreads stateless tasks
evenly, they add the same amount to
+ * every process's load and so cannot change the ranking at all. Where it
does not spread them evenly, only
+ * stateful work competes for the changelog reads a warm-up needs, so
counting stateless tasks would rank a
+ * process busy with work that does not compete as though it were a poor
place to restore.
+ *
+ * <p>A process running nothing but stateless tasks therefore has a load
of zero, which is the right answer
+ * here. That it holds no state to take over is beside the point: the
target assignment has already chosen every
+ * destination, and this order only decides which of those migrations is
funded first, never where a task goes.
+ *
+ * <p>Only {@link StreamsGroupMember#assignedTasks()} is counted -- {@link
+ * StreamsGroupMember#tasksPendingRevocation()} is deliberately not read,
and the two are disjoint, so nothing on
+ * its way out is counted. Counting a task the member has been told to
give up would overstate the load the
+ * process is about to carry, and would double-count the commonest shape
of all: a member being demoted from
+ * active to standby holds the task as a pending active revocation and as
an already-granted standby at once.
+ *
+ * @param members
+ * All members of the group.
+ * @param subtopologies
+ * The resolved subtopologies, which tell whether a subtopology is
stateful.
+ *
+ * @return The load of every process running at least one member, indexed
by process ID.
+ */
+ static Map<String, ProcessLoad> indexProcessLoad(
+ final Map<String, StreamsGroupMember> members,
+ final SortedMap<String, ConfiguredSubtopology> subtopologies
+ ) {
+ final Map<String, Integer> memberCounts = new HashMap<>();
+ final Map<String, Integer> statefulTaskCounts = new HashMap<>();
+
+ for (final StreamsGroupMember member : members.values()) {
+ final String processId = member.processId();
+ memberCounts.merge(processId, 1, Integer::sum);
+
+ final Consumer<TaskId> count = task ->
statefulTaskCounts.merge(processId, 1, Integer::sum);
+
forEachStatefulActiveTask(member.assignedTasks().activeTasksWithEpochs(),
subtopologies, count);
+ forEachStatefulTask(member.assignedTasks().standbyTasks(),
subtopologies, count);
+ forEachStatefulTask(member.assignedTasks().warmupTasks(),
subtopologies, count);
Review Comment:
Okay. So the increased load cannot really kick out a warm-up task that was
already created. It think it makes sense.
--
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]
