gabriellefu commented on code in PR #23482:
URL: https://github.com/apache/kafka/pull/23482#discussion_r4446136920
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -54,6 +54,15 @@ public class StickyTaskAssignor implements TaskAssignor {
Comparator.comparingInt((StandbyCandidate candidate) ->
candidate.isPrevStandby() ? 0 : 1)
.thenComparing(Comparator.comparingLong(StandbyCandidate::offsetSum).reversed());
+ /**
+ * Least loaded process first; among equally loaded processes, the one
with the fewest stateless active tasks, so
+ * that a stateless task fills up a process heavy on stateful ones.
Written as one lambda to keep heap operations cheap.
+ */
+ private static final Comparator<ProcessState> PROCESS_BY_LOAD = (process1,
process2) -> {
+ final int byLoad = Double.compare(process1.load(), process2.load());
+ return byLoad != 0 ? byLoad :
Double.compare(process1.statelessActiveLoad(), process2.statelessActiveLoad());
Review Comment:
thanks for the suggestion, i think it makes sense. I will open a follow up
pr to add the tie breaker
--
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]
