chia7712 commented on code in PR #23482:
URL: https://github.com/apache/kafka/pull/23482#discussion_r4438132418
##########
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:
Should we compare the `processId` as `BalancedTaskAssignor` does? Otherwise,
the order depends on the `HashMap` iteration order of `processIdToState` when
both `load` and `statelessActiveLoad` are equal, so the assignment is not
deterministic.
##########
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:
BTW, ProcessState has its own compareTo, but it seems to be unused. We will
file a patch to remove it.
```java
public int compareTo(final ProcessState other) {
int loadCompare = Double.compare(this.load, other.load());
if (loadCompare == 0) {
return Integer.compare(other.capacity, this.capacity);
}
return loadCompare;
}
```
--
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]
