lihaosky commented on code in PR #14178:
URL: https://github.com/apache/kafka/pull/14178#discussion_r1289205241
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentTestUtils.java:
##########
@@ -617,21 +652,23 @@ static Map<UUID, Map<String, Optional<String>>>
getRandomProcessRacks(final int
}
Collections.shuffle(racks);
Review Comment:
Looks `shuffle` can take random. I think we can use one static random in
this class
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java:
##########
@@ -505,4 +505,15 @@ private void assertNotAssigned(final TaskId task) {
throw new IllegalArgumentException("Tried to assign task " + task
+ ", but it is already assigned: " + this);
}
}
+
+ public ClientState copy() {
Review Comment:
Make sense. Let me add a copy constructor.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignor.java:
##########
@@ -346,6 +346,7 @@ public long optimizeActiveTasks(final SortedSet<TaskId>
activeTasks,
log.info("Assignment before active task optimization is {}\n with cost
{}", clientStates,
activeTasksCost(activeTasks, clientStates, trafficCost,
nonOverlapCost));
+ final long startTime = System.currentTimeMillis();
Review Comment:
There's a time in `StreamPartitionAssignor`, we can pass it into
`RackAwareTaskAssignor` constructor
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
