Repository: skiptests/astraea
Pull request #414 (2022-06-20 13:11:48)

Fix incorrect return type
如果是有額外考量才特地把 serializedKeySize 以及 serializedValueSize return type 設為 long 的話，歡迎關掉這份pr，感謝

brandboat: gentle ping @chia7712 

Patch:
diff --git a/app/src/main/java/org/astraea/app/performance/Metrics.java b/app/src/main/java/org/astraea/app/performance/Metrics.java
index a1cb8ff0..d0538149 100644
--- a/app/src/main/java/org/astraea/app/performance/Metrics.java
+++ b/app/src/main/java/org/astraea/app/performance/Metrics.java
@@ -19,7 +19,7 @@ package org.astraea.app.performance;
 import java.util.function.BiConsumer;
 
 /** Used to record statistics. This is thread safe. */
-public class Metrics implements BiConsumer<Long, Long> {
+public class Metrics implements BiConsumer<Long, Integer> {
   private double avgLatency;
   private long num;
   private long max;
@@ -38,11 +38,12 @@ public class Metrics implements BiConsumer<Long, Long> {
 
   /** Simultaneously add latency and bytes. */
   @Override
-  public synchronized void accept(Long latency, Long bytes) {
+  public synchronized void accept(Long latency, Integer bytes) {
     ++num;
     putLatency(latency);
     addBytes(bytes);
   }
+
   /** Add a new value to latency metric. */
   private synchronized void putLatency(long latency) {
     min = Math.min(min, latency);
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 afa52047..6021e900 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -97,7 +97,7 @@ public class Performance {
 
   static List<ProducerExecutor> producerExecutors(
       Performance.Argument argument,
-      List<? extends BiConsumer<Long, Long>> observers,
+      List<? extends BiConsumer<Long, Integer>> observers,
       DataSupplier dataSupplier,
       Supplier<Integer> partitionSupplier) {
     return IntStream.range(0, argument.producers)
@@ -211,7 +211,7 @@ public class Performance {
 
   static Executor consumerExecutor(
       Consumer<byte[], byte[]> consumer,
-      BiConsumer<Long, Long> observer,
+      BiConsumer<Long, Integer> observer,
       Manager manager,
       Supplier<Boolean> producerDone) {
     return new Executor() {
@@ -226,7 +226,7 @@ public class Performance {
                     // excluded)
                     observer.accept(
                         System.currentTimeMillis() - record.timestamp(),
-                        (long) record.serializedKeySize() + record.serializedValueSize());
+                        record.serializedKeySize() + record.serializedValueSize());
                   });
           // Consumer reached the record upperbound or consumed all the record producer produced.
           return producerDone.get() && manager.consumedDone() ? State.DONE : State.RUNNING;
diff --git a/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java b/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java
index 79d21470..371d4852 100644
--- a/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java
+++ b/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java
@@ -35,7 +35,7 @@ abstract class ProducerExecutor implements Executor {
       String topic,
       int batchSize,
       Producer<byte[], byte[]> producer,
-      BiConsumer<Long, Long> observer,
+      BiConsumer<Long, Integer> observer,
       Supplier<Integer> partitionSupplier,
       DataSupplier dataSupplier) {
     return new ProducerExecutor(topic, producer, partitionSupplier, observer, dataSupplier) {
@@ -95,7 +95,7 @@ abstract class ProducerExecutor implements Executor {
   static ProducerExecutor of(
       String topic,
       Producer<byte[], byte[]> producer,
-      BiConsumer<Long, Long> observer,
+      BiConsumer<Long, Integer> observer,
       Supplier<Integer> partitionSupplier,
       DataSupplier dataSupplier) {
     return new ProducerExecutor(topic, producer, partitionSupplier, observer, dataSupplier) {
@@ -139,14 +139,14 @@ abstract class ProducerExecutor implements Executor {
   private final String topic;
   private final Producer<byte[], byte[]> producer;
   private final Supplier<Integer> partitionSupplier;
-  private final BiConsumer<Long, Long> observer;
+  private final BiConsumer<Long, Integer> observer;
   private final DataSupplier dataSupplier;
 
   ProducerExecutor(
       String topic,
       Producer<byte[], byte[]> producer,
       Supplier<Integer> partitionSupplier,
-      BiConsumer<Long, Long> observer,
+      BiConsumer<Long, Integer> observer,
       DataSupplier dataSupplier) {
     this.topic = topic;
     this.producer = producer;
@@ -165,7 +165,7 @@ abstract class ProducerExecutor implements Executor {
     return partitionSupplier;
   }
 
-  BiConsumer<Long, Long> observer() {
+  BiConsumer<Long, Integer> observer() {
     return observer;
   }
 
diff --git a/app/src/main/java/org/astraea/app/producer/Metadata.java b/app/src/main/java/org/astraea/app/producer/Metadata.java
index aac52eca..4ee6ac81 100644
--- a/app/src/main/java/org/astraea/app/producer/Metadata.java
+++ b/app/src/main/java/org/astraea/app/producer/Metadata.java
@@ -65,7 +65,7 @@ public final class Metadata {
    * @return The size of the serialized, uncompressed key in bytes. If key is null, the returned
    *     size is -1.
    */
-  public long serializedKeySize() {
+  public int serializedKeySize() {
     return serializedKeySize;
   }
 
@@ -73,7 +73,7 @@ public final class Metadata {
    * @return The size of the serialized, uncompressed value in bytes. If value is null, the returned
    *     size is -1.
    */
-  public long serializedValueSize() {
+  public int serializedValueSize() {
     return serializedValueSize;
   }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/app/performance/ManagerTest.java b/app/src/test/java/org/astraea/app/performance/ManagerTest.java
index 8b068d7c..99e68545 100644
--- a/app/src/test/java/org/astraea/app/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/app/performance/ManagerTest.java
@@ -53,11 +53,11 @@ public class ManagerTest {
     var manager = new Manager(argument, List.of(producerMetrics), List.of(consumerMetrics));
 
     // Produce one record
-    producerMetrics.accept(0L, 0L);
+    producerMetrics.accept(0L, 0);
     Assertions.assertFalse(manager.consumedDone());
 
     // Consume one record
-    consumerMetrics.accept(0L, 0L);
+    consumerMetrics.accept(0L, 0);
     Assertions.assertTrue(manager.consumedDone());
 
     // Test zero consumer. (run for one record)
diff --git a/app/src/test/java/org/astraea/app/performance/MetricsTest.java b/app/src/test/java/org/astraea/app/performance/MetricsTest.java
index c9e5236c..eff94087 100644
--- a/app/src/test/java/org/astraea/app/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/app/performance/MetricsTest.java
@@ -34,7 +34,7 @@ public class MetricsTest {
     for (int i = 0; i < num; ++i) {
       long next = rand.nextInt();
       avg += ((double) next - avg) / (i + 1);
-      metrics.accept(next, 0L);
+      metrics.accept(next, 0);
     }
 
     Assertions.assertEquals(avg, metrics.avgLatency());
@@ -45,7 +45,7 @@ public class MetricsTest {
     var metrics = new Metrics();
 
     Assertions.assertEquals(0, metrics.bytes());
-    metrics.accept(0L, 1000L);
+    metrics.accept(0L, 1000);
     Assertions.assertEquals(1000, metrics.bytes());
   }
 
@@ -54,8 +54,8 @@ public class MetricsTest {
     var metrics = new Metrics();
 
     Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
-    metrics.accept(0L, 100L);
-    metrics.accept(0L, 101L);
+    metrics.accept(0L, 100);
+    metrics.accept(0L, 101);
     Assertions.assertEquals(201, metrics.clearAndGetCurrentBytes());
     Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
   }
diff --git a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
index 08e99a74..30222b51 100644
--- a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
@@ -45,7 +45,7 @@ public class PerformanceTest extends RequireBrokerCluster {
       "--bootstrap.servers", bootstrapServers(), "--topic", topic, "--transaction.size", "2"
     };
     var latch = new CountDownLatch(1);
-    BiConsumer<Long, Long> observer = (x, y) -> latch.countDown();
+    BiConsumer<Long, Integer> observer = (x, y) -> latch.countDown();
     var argument = Argument.parse(new Performance.Argument(), arguments1);
     var producerExecutors =
         Performance.producerExecutors(
@@ -67,7 +67,7 @@ public class PerformanceTest extends RequireBrokerCluster {
       "--bootstrap.servers", bootstrapServers(), "--topic", topic, "--compression", "gzip"
     };
     var latch = new CountDownLatch(1);
-    BiConsumer<Long, Long> observer = (x, y) -> latch.countDown();
+    BiConsumer<Long, Integer> observer = (x, y) -> latch.countDown();
     var argument = Argument.parse(new Performance.Argument(), arguments1);
     var producerExecutors =
         Performance.producerExecutors(
diff --git a/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java b/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java
index 2267c785..0b93d38e 100644
--- a/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java
+++ b/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java
@@ -97,12 +97,12 @@ public class ProducerExecutorTest extends RequireBrokerCluster {
         executor.transactional() ? 10 : 1, ((Observer) executor.observer()).elapsedHook.size());
   }
 
-  private static class Observer implements BiConsumer<Long, Long> {
+  private static class Observer implements BiConsumer<Long, Integer> {
     private final BlockingQueue<Long> recordsHook = new LinkedBlockingDeque<>();
-    private final BlockingQueue<Long> elapsedHook = new LinkedBlockingDeque<>();
+    private final BlockingQueue<Integer> elapsedHook = new LinkedBlockingDeque<>();
 
     @Override
-    public void accept(Long records, Long elapsed) {
+    public void accept(Long records, Integer elapsed) {
       Assertions.assertTrue(recordsHook.offer(records));
       Assertions.assertTrue(elapsedHook.offer(elapsed));
     }
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 7f723207..1a81c96d 100644
--- a/app/src/test/java/org/astraea/app/performance/TrackerTest.java
+++ b/app/src/test/java/org/astraea/app/performance/TrackerTest.java
@@ -36,8 +36,8 @@ public class TrackerTest {
     try (Tracker tracker = new Tracker(producerData, consumerData, manager, producerDone::get)) {
       producerDone.set(false);
       Assertions.assertEquals(State.RUNNING, tracker.execute());
-      producerData.get(0).accept(1L, 1L);
-      consumerData.get(0).accept(1L, 1L);
+      producerData.get(0).accept(1L, 1);
+      consumerData.get(0).accept(1L, 1);
       producerDone.set(true);
       Assertions.assertEquals(State.DONE, tracker.execute());
     }
@@ -48,7 +48,7 @@ public class TrackerTest {
     try (Tracker tracker = new Tracker(producerData, empty, manager, producerDone::get)) {
       producerDone.set(false);
       Assertions.assertEquals(State.RUNNING, tracker.execute());
-      producerData.get(0).accept(1L, 1L);
+      producerData.get(0).accept(1L, 1);
       producerDone.set(true);
       Assertions.assertEquals(State.DONE, tracker.execute());
     }
@@ -64,8 +64,8 @@ public class TrackerTest {
       Assertions.assertEquals(State.RUNNING, tracker.execute());
 
       // Mock record producing
-      producerData.get(0).accept(1L, 1L);
-      consumerData.get(0).accept(1L, 1L);
+      producerData.get(0).accept(1L, 1);
+      consumerData.get(0).accept(1L, 1);
       Thread.sleep(2000);
       producerDone.set(true);
       Assertions.assertEquals(State.DONE, tracker.execute());