Repository: skiptests/astraea
Pull request #212 (2022-02-07 14:43:46)

Fix consumerDone() false when no consumer
#209 

> `Manager#consumedDone` is always `false` when there is no consumer ...

Fixed by adding check on number of consumers.


Patch:
diff --git a/app/src/main/java/org/astraea/performance/Distribution.java b/app/src/main/java/org/astraea/performance/Distribution.java
index 8148aad6..bbfc5955 100644
--- a/app/src/main/java/org/astraea/performance/Distribution.java
+++ b/app/src/main/java/org/astraea/performance/Distribution.java
@@ -19,6 +19,8 @@ public interface Distribution {
       var name = rawArgument.split(":")[0];
 
       switch (name) {
+        case "fixed":
+          return fixed();
         case "uniform":
           if (args.length > 1) return uniform(Integer.parseInt(args[1]));
           return uniform();
@@ -34,6 +36,15 @@ public interface Distribution {
     }
   }
 
+  static Distribution fixed() {
+    var rand = new Random();
+    return fixed(rand.nextLong());
+  }
+
+  static Distribution fixed(long value) {
+    return () -> value;
+  }
+
   /** A distribution for providing a random long number from range [0, 2147483647) */
   static Distribution uniform() {
     return uniform(Integer.MAX_VALUE);
diff --git a/app/src/main/java/org/astraea/performance/FileWriter.java b/app/src/main/java/org/astraea/performance/FileWriter.java
index f59c00a8..f29e1756 100644
--- a/app/src/main/java/org/astraea/performance/FileWriter.java
+++ b/app/src/main/java/org/astraea/performance/FileWriter.java
@@ -6,6 +6,7 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.stream.IntStream;
 import org.astraea.concurrent.ThreadPool;
+import org.astraea.utils.DataUnit;
 
 public class FileWriter implements ThreadPool.Executor {
   private final Manager manager;
@@ -35,7 +36,7 @@ public class FileWriter implements ThreadPool.Executor {
     try {
       writer = new BufferedWriter(new java.io.FileWriter(CSVName));
       writer.write(
-          "Time \\ Name, Consumed/Produced, Output throughput (MiB/sec), Input throughput (MiB/sec), "
+          "Time \\ Name, Consumed/Produced, Output throughput (/sec), Input throughput (/sec), "
               + "Publish max latency (ms), Publish min latency (ms), "
               + "End-to-end max latency (ms), End-to-end min latency (ms)");
       IntStream.range(0, tracker.producerResult().bytes.size())
@@ -45,9 +46,9 @@ public class FileWriter implements ThreadPool.Executor {
                   writer.write(
                       ",Producer["
                           + i
-                          + "] average throughput (MB/sec), Producer["
+                          + "] current throughput (/sec), Producer["
                           + i
-                          + "] average publish latency (ms)");
+                          + "] current publish latency (ms)");
                 } catch (IOException ignore) {
                 }
               });
@@ -58,9 +59,9 @@ public class FileWriter implements ThreadPool.Executor {
                   writer.write(
                       ",Consumer["
                           + i
-                          + "] average throughput (MB/sec), Consumer["
+                          + "] current throughput (/sec), Consumer["
                           + i
-                          + "] average ene-to-end latency (ms)");
+                          + "] current ene-to-end latency (ms)");
                 } catch (IOException ignore) {
                 }
               });
@@ -89,16 +90,16 @@ public class FileWriter implements ThreadPool.Executor {
               + duration.toSecondsPart()
               + "s");
       writer.write(String.format(",%.2f%% / %.2f%%", consumerPercentage, producerPercentage));
-      writer.write("," + producerResult.averageBytes(duration));
-      writer.write("," + consumerResult.averageBytes(duration));
+      writer.write("," + DataUnit.Byte.of(producerResult.totalCurrentBytes()));
+      writer.write("," + DataUnit.Byte.of(consumerResult.totalCurrentBytes()));
       writer.write("," + producerResult.maxLatency + "," + producerResult.minLatency);
       writer.write("," + consumerResult.maxLatency + "," + consumerResult.minLatency);
       for (int i = 0; i < producerResult.bytes.size(); ++i) {
-        writer.write("," + Tracker.avg(duration, producerResult.bytes.get(i)));
+        writer.write("," + DataUnit.Byte.of(producerResult.currentBytes.get(i)));
         writer.write("," + producerResult.averageLatencies.get(i));
       }
       for (int i = 0; i < consumerResult.bytes.size(); ++i) {
-        writer.write("," + Tracker.avg(duration, consumerResult.bytes.get(i)));
+        writer.write("," + DataUnit.Byte.of(consumerResult.currentBytes.get(i)));
         writer.write("," + consumerResult.averageLatencies.get(i));
       }
       writer.newLine();
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index 2fb59b53..0b5fdec6 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -100,13 +100,13 @@ public class Manager {
 
   /** Check if we should keep consuming record. */
   public boolean consumedDone() {
-    return producedDone() && consumedRecords() >= producedRecords();
+    return producedDone()
+        && (consumerMetrics.size() == 0 || consumedRecords() >= producedRecords());
   }
 
   /** Randomly choose a key according to the distribution. */
-  public Optional<byte[]> getKey() {
-    if (distribution == null) return Optional.empty();
-    return Optional.of((String.valueOf(distribution.get())).getBytes());
+  public byte[] getKey() {
+    return (String.valueOf(distribution.get())).getBytes();
   }
 
   /** Randomly generate content before {@link #getContent()} is called. */
diff --git a/app/src/main/java/org/astraea/performance/Metrics.java b/app/src/main/java/org/astraea/performance/Metrics.java
index ec279c4f..90ebedbc 100644
--- a/app/src/main/java/org/astraea/performance/Metrics.java
+++ b/app/src/main/java/org/astraea/performance/Metrics.java
@@ -9,6 +9,7 @@ public class Metrics implements BiConsumer<Long, Long> {
   private long max;
   private long min;
   private long bytes;
+  private long currentBytes;
 
   public Metrics() {
     avgLatency = 0;
@@ -16,6 +17,7 @@ public class Metrics implements BiConsumer<Long, Long> {
     max = 0;
     min = Long.MAX_VALUE;
     bytes = 0;
+    currentBytes = 0;
   }
 
   /** Simultaneously add latency and bytes. */
@@ -33,6 +35,7 @@ public class Metrics implements BiConsumer<Long, Long> {
   }
   /** Add a new value to bytes. */
   private synchronized void addBytes(long bytes) {
+    this.currentBytes += bytes;
     this.bytes += bytes;
   }
 
@@ -57,4 +60,10 @@ public class Metrics implements BiConsumer<Long, Long> {
   public synchronized long bytes() {
     return bytes;
   }
+
+  public synchronized long clearAndGetCurrentBytes() {
+    var ans = currentBytes;
+    currentBytes = 0;
+    return ans;
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 5a4cd928..8846a60b 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -185,7 +185,7 @@ public class Performance {
             .sender()
             .topic(param.topic)
             .partition(partitions.get(rand.nextInt(partitions.size())))
-            .key(manager.getKey().orElse(null))
+            .key(manager.getKey())
             .value(payload.get())
             .timestamp(start)
             .run()
@@ -310,9 +310,9 @@ public class Performance {
     @Parameter(
         names = {"--key.distribution"},
         description =
-            "String: Distribution name. Available distribution names: \"uniform\", \"zipfian\", \"latest\". Default: (No key)",
+            "String: Distribution name. Available distribution names: \"fixed\" \"uniform\", \"zipfian\", \"latest\". Default: uniform",
         converter = Distribution.DistributionConverter.class)
-    Distribution distribution = null;
+    Distribution distribution = Distribution.uniform();
 
     @Parameter(
         names = {"--specify.broker"},
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index fbd25def..ab12bc1a 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -3,6 +3,7 @@ package org.astraea.performance;
 import java.time.Duration;
 import java.util.*;
 import org.astraea.concurrent.ThreadPool;
+import org.astraea.utils.DataUnit;
 
 /** Print out the given metrics. */
 public class Tracker implements ThreadPool.Executor {
@@ -60,12 +61,14 @@ public class Tracker implements ThreadPool.Executor {
             + duration.toSecondsPart()
             + "sec");
     System.out.printf("producers completion rate: %.2f%%%n", percentage);
-    System.out.printf("  Throughput: %.3fMB/second%n", result.averageBytes(duration));
+    System.out.printf("  average Throughput: %.3f MB/second%n", result.averageBytes(duration));
+    System.out.printf(
+        "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
     System.out.println("  publish max latency: " + result.maxLatency + "ms");
     System.out.println("  publish mim latency: " + result.minLatency + "ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
-          "  producer[%d] average throughput: %.3fMB%n", i, avg(duration, result.bytes.get(i)));
+          "  producer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
           "  producer[%d] average publish latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
@@ -82,12 +85,14 @@ public class Tracker implements ThreadPool.Executor {
     // Print out percentage of (consumed records) and (produced records)
     var percentage = result.completedRecords * 100D / manager.producedRecords();
     System.out.printf("consumer completion rate: %.2f%%%n", percentage);
-    System.out.printf("  throughput: %.3fMB/second%n", result.averageBytes(duration));
+    System.out.printf("  average throughput: %.3f MB/second%n", result.averageBytes(duration));
+    System.out.printf(
+        "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
     System.out.println("  end-to-end max latency: " + result.maxLatency + "ms");
     System.out.println("  end-to-end mim latency: " + result.minLatency + "ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
-          "  consumer[%d] average throughput: %.3fMB%n", i, avg(duration, result.bytes.get(i)));
+          "  consumer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
           "  consumer[%d] average ene-to-end latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
@@ -107,12 +112,14 @@ public class Tracker implements ThreadPool.Executor {
   private static Result result(List<Metrics> metrics) {
     var completed = 0;
     var bytes = new ArrayList<Long>();
+    var currentBytes = new ArrayList<Long>();
     var averageLatencies = new ArrayList<Double>();
     var max = 0L;
     var min = Long.MAX_VALUE;
     for (Metrics data : metrics) {
       completed += data.num();
       bytes.add(data.bytes());
+      currentBytes.add(data.clearAndGetCurrentBytes());
       averageLatencies.add(data.avgLatency());
       max = Math.max(max, data.max());
       min = Math.min(min, data.min());
@@ -120,6 +127,7 @@ public class Tracker implements ThreadPool.Executor {
     return new Result(
         completed,
         Collections.unmodifiableList(bytes),
+        Collections.unmodifiableList(currentBytes),
         Collections.unmodifiableList(averageLatencies),
         min,
         max);
@@ -128,6 +136,7 @@ public class Tracker implements ThreadPool.Executor {
   static class Result {
     public final long completedRecords;
     public final List<Long> bytes;
+    public final List<Long> currentBytes;
     public final List<Double> averageLatencies;
     public final long minLatency;
     public final long maxLatency;
@@ -135,11 +144,13 @@ public class Tracker implements ThreadPool.Executor {
     Result(
         long completedRecords,
         List<Long> bytes,
+        List<Long> currentBytes,
         List<Double> averageLatencies,
         long minLatency,
         long maxLatency) {
       this.completedRecords = completedRecords;
       this.bytes = bytes;
+      this.currentBytes = currentBytes;
       this.averageLatencies = averageLatencies;
       this.minLatency = minLatency;
       this.maxLatency = maxLatency;
@@ -152,5 +163,9 @@ public class Tracker implements ThreadPool.Executor {
     long totalBytes() {
       return bytes.stream().mapToLong(i -> i).sum();
     }
+
+    long totalCurrentBytes() {
+      return currentBytes.stream().mapToLong(i -> i).sum();
+    }
   }
 }
diff --git a/docker/start_kafka_tool.sh b/docker/start_kafka_tool.sh
index b51a116d..65997236 100755
--- a/docker/start_kafka_tool.sh
+++ b/docker/start_kafka_tool.sh
@@ -9,6 +9,10 @@ declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null
 declare -r DOCKERFILE=$DOCKER_FOLDER/kafka_tool.dockerfile
 declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
+declare -r ADDRESS=$([[ "$(which ipconfig)" != "" ]] && ipconfig getifaddr en0 || hostname -i)
+declare -r JMX_PORT=${JMX_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false \
+                     -Dcom.sun.management.jmxremote.port=$JMX_PORT -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT -Djava.rmi.server.hostname=$ADDRESS"
 
 # ===================================[functions]===================================
 
@@ -72,9 +76,11 @@ function buildImageIfNeed() {
 
 function runContainer() {
   local args=$1
+  echo "JMX address: $ADDRESS:$JMX_PORT"
   docker run --rm --init \
+    -p $JMX_PORT:$JMX_PORT \
     $IMAGE_NAME \
-    /bin/bash -c "java -jar /tmp/app.jar $args"
+    /bin/bash -c "java $JMX_OPTS -jar /tmp/app.jar $args"
 }
 
 # ===================================[main]===================================


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/DistributionTest.java b/app/src/test/java/org/astraea/performance/DistributionTest.java
index 2cb38401..6ff1e06f 100644
--- a/app/src/test/java/org/astraea/performance/DistributionTest.java
+++ b/app/src/test/java/org/astraea/performance/DistributionTest.java
@@ -1,9 +1,23 @@
 package org.astraea.performance;
 
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class DistributionTest {
+
+  @Test
+  void testFixed() {
+    var distribution = Distribution.fixed();
+    Assertions.assertEquals(
+        1,
+        IntStream.range(0, 10)
+            .mapToObj(ignored -> distribution.get())
+            .collect(Collectors.toSet())
+            .size());
+  }
+
   @Test
   void testLatest() throws InterruptedException {
     var distribution = Distribution.latest();
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index 57f72668..a6ef7934 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -60,18 +60,70 @@ public class ManagerTest {
 
     argument.distribution = Distribution.uniform();
     var manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isPresent());
+    Assertions.assertTrue(manager.getKey().length > 0);
 
     argument.distribution = Distribution.zipfian(10);
     manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isPresent());
+    Assertions.assertTrue(manager.getKey().length > 0);
 
     argument.distribution = Distribution.latest();
     manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isPresent());
+    Assertions.assertTrue(manager.getKey().length > 0);
 
-    argument.distribution = null;
+    argument.distribution = Distribution.fixed();
     manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isEmpty());
+    Assertions.assertTrue(manager.getKey().length > 0);
+  }
+
+  @Test
+  void testConsumerDone() {
+    var argument = new Performance.Argument();
+    var producerMetrics = List.of(new Metrics());
+    var consumerMetrics = List.of(new Metrics());
+
+    argument.exeTime = ExeTime.of("1records");
+    var manager = new Manager(argument, producerMetrics, consumerMetrics);
+    Assertions.assertFalse(manager.consumedDone());
+
+    // Produce one record
+    producerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.consumedDone());
+
+    // Consume one record
+    consumerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.consumedDone());
+
+    manager.producerClosed();
+    Assertions.assertTrue(manager.consumedDone());
+
+    // Test zero consumer. (run for one record)
+    producerMetrics = List.of(new Metrics());
+    consumerMetrics = List.of();
+    manager = new Manager(argument, producerMetrics, consumerMetrics);
+    Assertions.assertFalse(manager.consumedDone());
+
+    // Produce one record
+    producerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.consumedDone());
+
+    manager.producerClosed();
+    Assertions.assertTrue(manager.consumedDone());
+  }
+
+  @Test
+  void testProducerDone() {
+    var argument = new Performance.Argument();
+    argument.exeTime = ExeTime.of("1records");
+    var producerMetrics = List.of(new Metrics());
+    var consumerMetrics = List.of(new Metrics());
+    var manager = new Manager(new Performance.Argument(), producerMetrics, consumerMetrics);
+
+    Assertions.assertFalse(manager.producedDone());
+
+    producerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.producedDone());
+
+    manager.producerClosed();
+    Assertions.assertTrue(manager.producedDone());
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/MetricsTest.java b/app/src/test/java/org/astraea/performance/MetricsTest.java
index ac9799da..e2c23a78 100644
--- a/app/src/test/java/org/astraea/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/performance/MetricsTest.java
@@ -32,4 +32,15 @@ public class MetricsTest {
     metrics.accept(0L, 1000L);
     Assertions.assertEquals(1000, metrics.bytes());
   }
+
+  @Test
+  void testCurrentBytes() {
+    var metrics = new Metrics();
+
+    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
+    metrics.accept(0L, 100L);
+    metrics.accept(0L, 101L);
+    Assertions.assertEquals(201, metrics.clearAndGetCurrentBytes());
+    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
+  }
 }