[PERF] Enhance performance partition change rate

app/src/main/java/org/astraea/app/performance/ConsumerThread.java

@@ -37,7 +37,10 @@
 
 public interface ConsumerThread extends AbstractThread {
 
-  ConcurrentMap<String, Set<TopicPartition>> CLIENT_ID_PARTITIONS = new ConcurrentHashMap<>();
+  ConcurrentMap<String, Set<TopicPartition>> CLIENT_ID_ASSIGNED_PARTITIONS =
+      new ConcurrentHashMap<>();
+  ConcurrentMap<String, Set<TopicPartition>> CLIENT_ID_REVOKED_PARTITIONS =
+      new ConcurrentHashMap<>();
 
   static List<ConsumerThread> create(
       int consumers,
@@ -64,8 +67,7 @@ static List<ConsumerThread> create(
             index -> {
               @SuppressWarnings("resource")
               var clientId = Utils.randomString();
-              var consumer =
-                  consumerSupplier.apply(clientId, ps -> CLIENT_ID_PARTITIONS.put(clientId, ps));
+              var consumer = consumerSupplier.apply(clientId, new PartitionRatioListener(clientId));
               var closed = new AtomicBoolean(false);
               var closeLatch = closeLatches.get(index);
               var subscribed = new AtomicBoolean(true);
@@ -87,7 +89,8 @@ static List<ConsumerThread> create(
                       Utils.swallowException(consumer::close);
                       closeLatch.countDown();
                       closed.set(true);
-                      CLIENT_ID_PARTITIONS.remove(clientId);
+                      CLIENT_ID_ASSIGNED_PARTITIONS.remove(clientId);
+                      CLIENT_ID_REVOKED_PARTITIONS.remove(clientId);
                     }
                   });
               return new ConsumerThread() {

app/src/main/java/org/astraea/app/performance/PartitionRatioListener.java

@@ -0,0 +1,40 @@
+/*
+ * 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.astraea.app.performance;
+
+import java.util.Set;
+import org.astraea.common.admin.TopicPartition;
+import org.astraea.common.consumer.ConsumerRebalanceListener;
+
+public class PartitionRatioListener implements ConsumerRebalanceListener {
+  private final String clientId;
+
+  PartitionRatioListener(String clientId) {
+    this.clientId = clientId;
+  }
+
+  @Override
+  public void onPartitionAssigned(Set<TopicPartition> partitions) {
+    ConsumerThread.CLIENT_ID_REVOKED_PARTITIONS.putIfAbsent(clientId, Set.of());
+    ConsumerThread.CLIENT_ID_ASSIGNED_PARTITIONS.put(clientId, partitions);
+  }
+
+  @Override
+  public void onPartitionsRevoked(Set<TopicPartition> partitions) {
+    ConsumerThread.CLIENT_ID_REVOKED_PARTITIONS.put(clientId, partitions);
+  }
+}

app/src/main/java/org/astraea/app/performance/ReportFormat.java

@@ -30,6 +30,7 @@
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
 import org.astraea.common.EnumInfo;
@@ -207,6 +208,17 @@ private static List<CSVContentElement> latencyAndIO() {
                   CSVContentElement.create(
                       "Consumer[" + i + "] average publish latency (ms)",
                       () -> Double.toString(consumerReports.get(i).avgLatency())));
+              elements.add(
+                  CSVContentElement.create(
+                      "Consumer[" + i + "] partition difference",
+                      () ->
+                          Integer.toString(
+                              (ConsumerThread.CLIENT_ID_ASSIGNED_PARTITIONS
+                                      .getOrDefault(consumerReports.get(i).clientId(), Set.of())
+                                      .size()
+                                  - ConsumerThread.CLIENT_ID_REVOKED_PARTITIONS
+                                      .getOrDefault(consumerReports.get(i).clientId(), Set.of())
+                                      .size()))));
             });
     return elements;
   }

app/src/main/java/org/astraea/app/performance/TrackerThread.java

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -140,10 +141,17 @@ boolean tryToPrint(Duration duration) {
           .ifPresent(i -> System.out.printf("  rebalance average latency: %.3f ms%n", i));
       for (var i = 0; i < reports.size(); ++i) {
         var report = reports.get(i);
+        var clientId = report.clientId();
         var ms = metrics.stream().filter(m -> m.clientId().equals(report.clientId())).findFirst();
         if (ms.isPresent()) {
           System.out.printf(
-              "  consumer[%d] has %d partitions%n", i, (int) ms.get().assignedPartitions());
+              "  consumer[%d] has %d partitions and its partition assign difference is %d %n",
+              i,
+              (int) ms.get().assignedPartitions(),
+              (ConsumerThread.CLIENT_ID_ASSIGNED_PARTITIONS.getOrDefault(clientId, Set.of()).size()
+                  - ConsumerThread.CLIENT_ID_REVOKED_PARTITIONS
+                      .getOrDefault(clientId, Set.of())
+                      .size()));
         }
         System.out.printf(
             "  consumed[%d] average throughput: %s%n",