KAFKA-17710; Rework uniform heterogeneous assignor to improve perf (#17385)

Rework the uniform heterogeneous assignor to improve performance, while
preserving the high level ideas and structure from the existing
implementation:
 * The assignor works in 3 stages: importing the previous assignment for
   stickiness, assigning unassigned partitions and iteratively
   reassigning partitions to improve balance.
 * Unassigned partitions are assigned to the subscribers with the least
   number of partitions. This maximizes balance within a single topic.
 * During the iterative rebalancing phase, partitions are reassigned to
   their previous owner if it improves balance (stickiness restoration).
 * During the iterative rebalancing phase, partitions are reassigned to
   the subscriber with the least number of partitions to improve
   balance.

A non-exhaustive list of changes is:
 * The assignment of unassigned partitions and iterative reassignment
   stages now works through partitions topic by topic. Previously
   partitions from topics with the same number of partitions per
   subscriber would be interleaved. Since we iterate topic by topic, we
   can reuse data about topic subscribers.
 * Instead of maintaining TreeSets to find the least loaded subscribers,
   we sort an ArrayList of subscribers once per topic and start filling
   up subscribers, least loaded first. In testing, this approach was
   found to be faster than maintaining PriorityQueues.
 * Implement stickiness restoration by creating a mapping of partitions
   to previous owner and checking against that mapping, instead of
   tracking partition movements during iterative reassignment.
 * Track member partition counts using a plain int array, to avoid
   overhead from boxing and HashMap lookups. Member partition counts are
   accessed very frequently and this needs to be fast. As a consequence,
   we have to number members 0 to M - 1.
 * Bound the iterative reassignment stage to a fixed number of
   iterations. Under some uncommon subscription patterns, the iterative
   reassignment stage converges slowly. In these cases, the iterative
   reassignment stage terminates without producing an optimally balanced
   assignment anyway (see javadoc for balanceTopics).
 * Re-use Maps from the previous assignment where possible,
   ie. introduce a copy-on-write mechanism while computing the new
   assignment.

Reviewers: David Jacot <djacot@confluent.io>

Author: squah-confluent

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignorHelpers.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.assignor;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Provides helper methods for assignors.
+ */
+public final class AssignorHelpers {
+    private static final Class<?> UNMODIFIABLE_MAP_CLASS = Collections.unmodifiableMap(new HashMap<>()).getClass();
+    private static final Class<?> EMPTY_MAP_CLASS = Collections.emptyMap().getClass();
+
+    /**
+     * @return True if the provided map is an UnmodifiableMap or EmptyMap. Those classes are not
+     * public hence we cannot use the `instanceof` operator.
+     */
+    public static boolean isImmutableMap(Map<?, ?> map) {
+        return UNMODIFIABLE_MAP_CLASS.isInstance(map) || EMPTY_MAP_CLASS.isInstance(map);
+    }
+
+    /**
+     * Deep copies a member assignment map.
+     * @param map The assignment to copy.
+     * @return A deep copy of the assignment.
+     */
+    public static Map<Uuid, Set<Integer>> deepCopyAssignment(Map<Uuid, Set<Integer>> map) {
+        Map<Uuid, Set<Integer>> copy = new HashMap<>(map.size());
+        for (Map.Entry<Uuid, Set<Integer>> entry : map.entrySet()) {
+            copy.put(entry.getKey(), new HashSet<>(entry.getValue()));
+        }
+        return copy;
+    }
+}