mjsax commented on code in PR #24090:
URL: https://github.com/apache/kafka/pull/24090#discussion_r4499299381
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -57,11 +57,16 @@ public class StickyTaskAssignor implements TaskAssignor {
/**
* 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.
+ * that a stateless task fills up a process heavy on stateful ones; then
by process id, so the order is deterministic.
+ * 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());
+ if (byLoad != 0) {
+ return byLoad;
+ }
+ final int byStatelessActiveLoad =
Double.compare(process1.statelessActiveLoad(), process2.statelessActiveLoad());
+ return byStatelessActiveLoad != 0 ? byStatelessActiveLoad :
process1.processId().compareTo(process2.processId());
Review Comment:
Should we rewrite this as:
```
private static final Comparator<ProcessTasks> BY_ASSIGNED_LOAD =
Comparator.comparingDouble(ProcessState::load)
.thenComparing(ProcessState::statelessActiveLoad)
.thenComparing(ProcessState::processId);
```
--
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]
