Repository: skiptests/astraea
Pull request #602 (2022-08-20 14:41:50)

Performance shows the incorrect producer info after all producers are…
兩個修正：

1. 避免使用固定的reports，這樣方便動態調整數量
2. 當producer/consumer完成後就不再輸出狀態


Patch:
diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
index 1050ae01..1ad2c557 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -137,16 +137,18 @@ public class Performance {
                     .consumerRebalanceListener(listener)
                     .build());
 
-    var producerReports =
-        producerThreads.stream()
-            .map(ProducerThread::report)
-            .collect(Collectors.toUnmodifiableList());
-    var consumerReports =
-        consumerThreads.stream()
-            .map(ConsumerThread::report)
-            .collect(Collectors.toUnmodifiableList());
-
-    var tracker = TrackerThread.create(producerReports, consumerReports, param.exeTime);
+    Supplier<List<ProducerThread.Report>> producerReporter =
+        () ->
+            producerThreads.stream()
+                .map(ProducerThread::report)
+                .collect(Collectors.toUnmodifiableList());
+    Supplier<List<ConsumerThread.Report>> consumerReporter =
+        () ->
+            consumerThreads.stream()
+                .map(ConsumerThread::report)
+                .collect(Collectors.toUnmodifiableList());
+
+    var tracker = TrackerThread.create(producerReporter, consumerReporter, param.exeTime);
 
     Optional<Runnable> fileWriter =
         param.CSVPath == null
@@ -157,8 +159,8 @@ public class Performance {
                     param.CSVPath,
                     () -> consumerThreads.stream().allMatch(AbstractThread::closed),
                     () -> producerThreads.stream().allMatch(AbstractThread::closed),
-                    producerReports,
-                    consumerReports));
+                    producerReporter,
+                    consumerReporter));
 
     var fileWriterFuture =
         fileWriter.map(CompletableFuture::runAsync).orElse(CompletableFuture.completedFuture(null));
@@ -190,7 +192,7 @@ public class Performance {
             if (offsets.entrySet().stream()
                 .allMatch(
                     e ->
-                        consumerReports.stream()
+                        consumerReporter.get().stream()
                             .anyMatch(r -> r.offset(e.getKey()) >= e.getValue()))) break;
             Utils.sleep(Duration.ofSeconds(2));
           }
diff --git a/app/src/main/java/org/astraea/app/performance/ReportFormat.java b/app/src/main/java/org/astraea/app/performance/ReportFormat.java
index 5e3b80d4..83c7fdf3 100644
--- a/app/src/main/java/org/astraea/app/performance/ReportFormat.java
+++ b/app/src/main/java/org/astraea/app/performance/ReportFormat.java
@@ -63,8 +63,8 @@ public enum ReportFormat {
       Path path,
       Supplier<Boolean> consumerDone,
       Supplier<Boolean> producerDone,
-      List<ProducerThread.Report> producerReports,
-      List<ConsumerThread.Report> consumerReports)
+      Supplier<List<ProducerThread.Report>> producerReporter,
+      Supplier<List<ConsumerThread.Report>> consumerReporter)
       throws IOException {
     var filePath =
         FileSystems.getDefault()
@@ -75,7 +75,7 @@ public enum ReportFormat {
                     + "."
                     + reportFormat);
     var writer = new BufferedWriter(new FileWriter(filePath.toFile()));
-    var elements = latencyAndIO(producerReports, consumerReports);
+    var elements = latencyAndIO(producerReporter, consumerReporter);
     switch (reportFormat) {
       case CSV:
         initCSVFormat(writer, elements);
@@ -156,7 +156,10 @@ public enum ReportFormat {
   }
 
   private static List<CSVContentElement> latencyAndIO(
-      List<ProducerThread.Report> producerReports, List<ConsumerThread.Report> consumerReports) {
+      Supplier<List<ProducerThread.Report>> producerReporter,
+      Supplier<List<ConsumerThread.Report>> consumerReporter) {
+    var producerReports = producerReporter.get();
+    var consumerReports = consumerReporter.get();
     var elements = new ArrayList<CSVContentElement>();
     elements.add(
         CSVContentElement.create(
diff --git a/app/src/main/java/org/astraea/app/performance/TrackerThread.java b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
index 4dc4e3b8..96284698 100644
--- a/app/src/main/java/org/astraea/app/performance/TrackerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
@@ -21,6 +21,7 @@ import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import org.astraea.app.common.Utils;
 
@@ -28,28 +29,19 @@ import org.astraea.app.common.Utils;
 public interface TrackerThread extends AbstractThread {
 
   static TrackerThread create(
-      List<ProducerThread.Report> producerReports,
-      List<ConsumerThread.Report> consumerReports,
+      Supplier<List<ProducerThread.Report>> producerReporter,
+      Supplier<List<ConsumerThread.Report>> consumerReporter,
       ExeTime exeTime) {
     var start = System.currentTimeMillis() - Duration.ofSeconds(1).toMillis();
 
-    Supplier<Boolean> logProducers =
-        () -> {
+    Function<Duration, Boolean> logProducers =
+        duration -> {
+          var producerReports = producerReporter.get();
           var records = producerReports.stream().mapToLong(Report::records).sum();
           if (records == 0) return false;
-
-          var duration = Duration.ofMillis(System.currentTimeMillis() - start);
           // Print completion rate (by number of records) or (by time)
           var percentage = Math.min(100D, exeTime.percentage(records, duration.toMillis()));
 
-          System.out.println(
-              "Time: "
-                  + duration.toHoursPart()
-                  + "hr "
-                  + duration.toMinutesPart()
-                  + "min "
-                  + duration.toSecondsPart()
-                  + "sec");
           System.out.printf("producers completion rate: %.2f%%%n", percentage);
           System.out.printf(
               "  average throughput: %.3f MB/second%n",
@@ -75,15 +67,15 @@ public interface TrackerThread extends AbstractThread {
           return percentage >= 100;
         };
 
-    Supplier<Boolean> logConsumers =
-        () -> {
+    Function<Duration, Boolean> logConsumers =
+        duration -> {
+          var producerReports = producerReporter.get();
+          var consumerReports = consumerReporter.get();
           // there is no consumer, so we just complete this log.
           if (consumerReports.isEmpty()) return true;
 
           if (consumerReports.stream().mapToLong(Report::records).sum() == 0) return false;
 
-          var duration = Duration.ofMillis(System.currentTimeMillis() - start);
-
           // Print out percentage of (consumed records) and (produced records)
           var producerOffset =
               Report.maxOffsets(producerReports).values().stream().mapToLong(v -> v).sum();
@@ -131,8 +123,21 @@ public interface TrackerThread extends AbstractThread {
     CompletableFuture.runAsync(
         () -> {
           try {
+            var producerDone = false;
+            var consumerDone = false;
             while (!closed.get()) {
-              if (logProducers.get() & logConsumers.get()) return;
+              var duration = Duration.ofMillis(System.currentTimeMillis() - start);
+              System.out.println(
+                  "Time: "
+                      + duration.toHoursPart()
+                      + "hr "
+                      + duration.toMinutesPart()
+                      + "min "
+                      + duration.toSecondsPart()
+                      + "sec");
+              if (!producerDone) producerDone = logProducers.apply(duration);
+              if (!consumerDone) consumerDone = logConsumers.apply(duration);
+              if (producerDone && consumerDone) return;
 
               // Log after waiting for one second
               Utils.sleep(Duration.ofSeconds(1));


Test patch:
diff --git a/app/src/test/java/org/astraea/app/performance/TrackerTest.java b/app/src/test/java/org/astraea/app/performance/TrackerTest.java
index 1cb4ae8c..aca253c7 100644
--- a/app/src/test/java/org/astraea/app/performance/TrackerTest.java
+++ b/app/src/test/java/org/astraea/app/performance/TrackerTest.java
@@ -26,7 +26,7 @@ public class TrackerTest {
 
   @Test
   void testClose() {
-    var tracker = TrackerThread.create(List.of(), List.of(), ExeTime.of("1records"));
+    var tracker = TrackerThread.create(() -> List.of(), () -> List.of(), ExeTime.of("1records"));
     Assertions.assertFalse(tracker.closed());
     tracker.close();
     Assertions.assertTrue(tracker.closed());
@@ -35,7 +35,9 @@ public class TrackerTest {
   @Test
   void testZeroConsumer() {
     var producerReport = new ProducerThread.Report();
-    var tracker = TrackerThread.create(List.of(producerReport), List.of(), ExeTime.of("1records"));
+    var tracker =
+        TrackerThread.create(
+            () -> List.of(producerReport), () -> List.of(), ExeTime.of("1records"));
     Assertions.assertFalse(tracker.closed());
     producerReport.record("topic", 1, 100, 1L, 1);
     // wait to done
@@ -48,7 +50,8 @@ public class TrackerTest {
     var producerReport = new ProducerThread.Report();
     var consumerReport = new ConsumerThread.Report();
     var tracker =
-        TrackerThread.create(List.of(producerReport), List.of(consumerReport), ExeTime.of("2s"));
+        TrackerThread.create(
+            () -> List.of(producerReport), () -> List.of(consumerReport), ExeTime.of("2s"));
     Assertions.assertFalse(tracker.closed());
     producerReport.record("topic", 1, 100, 1L, 1);
     consumerReport.record("topic", 1, 100, 1L, 1);
@@ -62,7 +65,7 @@ public class TrackerTest {
     var consumerReport = new ConsumerThread.Report();
     var tracker =
         TrackerThread.create(
-            List.of(producerReport), List.of(consumerReport), ExeTime.of("1records"));
+            () -> List.of(producerReport), () -> List.of(consumerReport), ExeTime.of("1records"));
     Assertions.assertFalse(tracker.closed());
     producerReport.record("topic", 1, 100, 1L, 1);
     consumerReport.record("topic", 1, 100, 1L, 1);