Repository: skiptests/astraea
Pull request #151 (2021-12-15 10:21:19)

修改TopicAdmin interface
為函式`topics()`、`topicNames()`新增一個參數，用來決定是否要抓取kafka內部使用的topic。


Patch:
diff --git a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
index 9e0bcccd..dac0c30f 100644
--- a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
@@ -4,11 +4,10 @@ import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiFunction;
 import java.util.function.Function;
-import java.util.function.Supplier;
 import org.astraea.Utils;
 import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.jmx.MBeanClient;
@@ -51,7 +50,7 @@ public class BeanCollector {
   private final int numberOfObjectsPerNode;
 
   // visible for testing
-  final ConcurrentMap<Node, NodeClient> clients = new ConcurrentSkipListMap<>();
+  final ConcurrentMap<String, Node> nodes = new ConcurrentSkipListMap<>();
 
   private BeanCollector(
       BiFunction<String, Integer, MBeanClient> clientCreator,
@@ -88,10 +87,12 @@ public class BeanCollector {
 
       @Override
       public Receiver build() {
-        var node = new Node(host, port);
-        var client = clients.computeIfAbsent(node, ignored -> new NodeClient());
-        return client.add(
+        var nodeKey = host + ":" + port;
+        var node = nodes.computeIfAbsent(nodeKey, ignored -> new Node(host, port));
+        var receiver =
             new Receiver() {
+              private final Map<Long, HasBeanObject> objects = new HashMap<>();
+
               @Override
               public String host() {
                 return host;
@@ -104,92 +105,60 @@ public class BeanCollector {
 
               @Override
               public List<HasBeanObject> current() {
-                client.tryUpdate(
-                    interval,
-                    numberOfObjectsPerNode,
-                    getter,
-                    () -> clientCreator.apply(host, port));
-                return client.current();
+                tryUpdate();
+                return List.copyOf(objects.values());
               }
 
               @Override
               public void close() {
-                client.remove(this);
+                node.lock.lock();
+                try {
+                  node.receivers.remove(this);
+                  if (node.receivers.isEmpty()) Utils.close(node.mBeanClient);
+                  node.mBeanClient = null;
+                } finally {
+                  node.lock.unlock();
+                }
               }
-            });
-      }
-    };
-  }
 
-  // visible for testing
-  static class NodeClient {
-    private final Set<Receiver> receivers = new HashSet<>();
-    private final ReadWriteLock lock = new ReentrantReadWriteLock();
-    // visible for testing
-    MBeanClient mBeanClient;
-    private final Map<Long, HasBeanObject> objects = new HashMap<>();
-
-    Receiver add(Receiver receiver) {
-      lock.writeLock().lock();
-      try {
-        receivers.add(receiver);
-      } finally {
-        lock.writeLock().unlock();
-      }
-      return receiver;
-    }
-
-    void remove(Receiver receiver) {
-      lock.writeLock().lock();
-      try {
-        receivers.remove(receiver);
-        if (receivers.isEmpty()) Utils.close(mBeanClient);
-        mBeanClient = null;
-      } finally {
-        lock.writeLock().unlock();
-      }
-    }
-
-    private void removeOldest() {
-      objects.keySet().stream().min((Long::compare)).ifPresent(objects::remove);
-    }
-
-    List<HasBeanObject> current() {
-      lock.readLock().lock();
-      try {
-        return List.copyOf(objects.values());
-      } finally {
-        lock.readLock().unlock();
-      }
-    }
-
-    void tryUpdate(
-        Duration interval,
-        int numberOfObjectsPerNode,
-        Function<MBeanClient, HasBeanObject> beanGetter,
-        Supplier<MBeanClient> clientCreator) {
+              private void tryUpdate() {
+                var needUpdate =
+                    objects.keySet().stream()
+                        .max((Long::compare))
+                        .map(last -> last + interval.toMillis() <= System.currentTimeMillis())
+                        .orElse(true);
+                if (needUpdate && node.lock.tryLock()) {
+                  try {
+                    if (node.mBeanClient == null)
+                      node.mBeanClient = clientCreator.apply(host, port);
+                    if (objects.size() >= numberOfObjectsPerNode)
+                      objects.keySet().stream().min((Long::compare)).ifPresent(objects::remove);
+                    objects.put(System.currentTimeMillis(), getter.apply(node.mBeanClient));
+                  } finally {
+                    node.lock.unlock();
+                  }
+                }
+              }
+            };
 
-      if (lock.writeLock().tryLock()) {
+        // add receiver
+        node.lock.lock();
         try {
-          var needUpdate =
-              objects.keySet().stream()
-                  .max((Long::compare))
-                  .map(last -> last + interval.toMillis() <= System.currentTimeMillis())
-                  .orElse(true);
-          if (needUpdate) {
-            if (mBeanClient == null) mBeanClient = clientCreator.get();
-            if (objects.size() >= numberOfObjectsPerNode) removeOldest();
-            objects.put(System.currentTimeMillis(), beanGetter.apply(mBeanClient));
-          }
-
+          node.receivers.add(receiver);
         } finally {
-          lock.writeLock().unlock();
+          node.lock.unlock();
         }
+        return receiver;
       }
-    }
+    };
   }
 
-  private static final class Node implements Comparable<Node> {
+  // visible for testing
+  static final class Node {
+    private final Set<Receiver> receivers = new HashSet<>();
+    private final Lock lock = new ReentrantLock();
+    // visible for testing
+    MBeanClient mBeanClient;
     public final String host;
     public final int port;
 
@@ -197,12 +166,5 @@ public class BeanCollector {
       this.host = host;
       this.port = port;
     }
-
-    @Override
-    public int compareTo(Node other) {
-      var result = host.compareTo(other.host);
-      if (result != 0) return result;
-      return Integer.compare(port, other.port);
-    }
   }
 }
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index acfbb513..dc835dce 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -163,9 +163,22 @@ public class Builder {
 
     @Override
     public Map<String, TopicConfig> topics() {
+      return topics(true);
+    }
+
+    @Override
+    public Map<String, TopicConfig> publicTopics() {
+      return topics(false);
+    }
+
+    private Map<String, TopicConfig> topics(boolean listInternal) {
       var topics =
           Utils.handleException(
-              () -> admin.listTopics(new ListTopicsOptions().listInternal(true)).names().get());
+              () ->
+                  admin
+                      .listTopics(new ListTopicsOptions().listInternal(listInternal))
+                      .names()
+                      .get());
       return Utils.handleException(
               () ->
                   admin
diff --git a/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java b/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java
index 784aeb75..77d9bb72 100644
--- a/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java
+++ b/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java
@@ -4,6 +4,7 @@ import com.beust.jcommander.Parameter;
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -14,6 +15,9 @@ import org.apache.kafka.common.TopicPartitionReplica;
 import org.astraea.Utils;
 import org.astraea.argument.ArgumentUtil;
 import org.astraea.argument.BasicArgumentWithPropFile;
+import org.astraea.utils.DataRate;
+import org.astraea.utils.DataSize;
+import org.astraea.utils.DataUnit;
 
 public class ReplicaSyncingMonitor {
 
@@ -96,11 +100,12 @@ public class ReplicaSyncingMonitor {
                                   entry ->
                                       Map.entry(
                                           entry.getValue(),
-                                          new DataRate(
-                                              leaderReplica.size(),
-                                              previousCheckedSize.getOrDefault(
-                                                  entry.getKey(), entry.getValue().size()),
-                                              entry.getValue().size(),
+                                          new ProgressInfo(
+                                              DataUnit.Byte.of(leaderReplica.size()),
+                                              DataUnit.Byte.of(
+                                                  previousCheckedSize.getOrDefault(
+                                                      entry.getKey(), entry.getValue().size())),
+                                              DataUnit.Byte.of(entry.getValue().size()),
                                               argument.interval)))
                               .map(
                                   entry ->
@@ -169,7 +174,7 @@ public class ReplicaSyncingMonitor {
     }
   }
 
-  private static String formatString(Replica replica, DataRate value) {
+  private static String formatString(Replica replica, ProgressInfo value) {
     return Stream.of(
             Optional.ofNullable(value.progressBar()),
             Optional.ofNullable(replica.leader() ? null : value.dataRateString()),
@@ -190,14 +195,15 @@ public class ReplicaSyncingMonitor {
         .collect(Collectors.toSet());
   }
 
-  static class DataRate {
-    public final long leaderSize;
-    public final long previousSize;
-    public final long currentSize;
-    public final Duration interval;
+  static class ProgressInfo {
+    private final DataSize leaderSize;
+    private final DataSize previousSize;
+    private final DataSize currentSize;
+    private final Duration interval;
 
-    DataRate(long leaderSize, long previousSize, long currentSize, Duration interval) {
-      if (previousSize > currentSize) throw new IllegalArgumentException();
+    ProgressInfo(
+        DataSize leaderSize, DataSize previousSize, DataSize currentSize, Duration interval) {
+      if (previousSize.greaterThan(leaderSize)) throw new IllegalArgumentException();
       this.leaderSize = leaderSize;
       this.previousSize = previousSize;
       this.currentSize = currentSize;
@@ -205,7 +211,9 @@ public class ReplicaSyncingMonitor {
     }
 
     public double progress() {
-      return ((double) currentSize) / leaderSize * 100.0;
+      var currentBit = currentSize.measurement(DataUnit.Bit).doubleValue();
+      var leaderBit = leaderSize.measurement(DataUnit.Bit).doubleValue();
+      return currentBit / leaderBit * 100.0;
     }
 
     public String progressBar() {
@@ -221,8 +229,16 @@ public class ReplicaSyncingMonitor {
           progress());
     }
 
-    public double dataRatePerSec() {
-      return (double) (currentSize - previousSize) / ((interval.toNanos() / 1e9));
+    public DataRate dataRate() {
+      return currentSize.subtract(previousSize).dataRate(interval);
+    }
+
+    public double dataRate(DataUnit dataUnit, ChronoUnit chronoUnit) {
+      return dataRate().toBigDecimal(dataUnit, chronoUnit).doubleValue();
+    }
+
+    public boolean isStalled() {
+      return currentSize.equals(previousSize);
     }
 
     /**
@@ -232,8 +248,14 @@ public class ReplicaSyncingMonitor {
      *     stalled.
      */
     public Duration estimateFinishTime() {
-      if (dataRatePerSec() == 0) return Duration.ofSeconds(-1);
-      return Duration.ofSeconds((long) ((leaderSize - currentSize) / dataRatePerSec()));
+      if (isStalled()) return Duration.ofSeconds(-1);
+      else {
+        final double leaderByte = leaderSize.measurement(DataUnit.Byte).doubleValue();
+        final double currentByte = currentSize.measurement(DataUnit.Byte).doubleValue();
+        final double estimateFinishTime =
+            ((leaderByte - currentByte) / dataRate(DataUnit.Byte, ChronoUnit.SECONDS));
+        return Duration.ofSeconds((long) estimateFinishTime);
+      }
     }
 
     public String estimateFinishTimeString() {
@@ -252,21 +274,12 @@ public class ReplicaSyncingMonitor {
     }
 
     public String dataRateString() {
-      final double sizeProgress = dataRatePerSec();
-      final long TB = 1024L * 1024L * 1024L * 1024L;
-      final long GB = 1024L * 1024L * 1024L;
-      final long MB = 1024L * 1024L;
-      final long KB = 1024L;
-      if (sizeProgress > TB) return String.format("%.2f TB/s", sizeProgress / TB);
-      else if (sizeProgress > GB) return String.format("%.2f GB/s", sizeProgress / GB);
-      else if (sizeProgress > MB) return String.format("%.2f MB/s", sizeProgress / MB);
-      else if (sizeProgress > KB) return String.format("%.2f KB/s", sizeProgress / KB);
-      else return String.format("%.2f B/s", sizeProgress);
+      return dataRate().toString(ChronoUnit.SECONDS);
     }
 
     @Override
     public String toString() {
-      if (estimateFinishTime().isZero() && currentSize == previousSize) return dataRateString();
+      if (estimateFinishTime().isZero() && isStalled()) return dataRateString();
       else return String.format("%s (%s)", dataRateString(), estimateFinishTimeString());
     }
   }
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index d11434a2..be815c56 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -26,6 +26,13 @@ public interface TopicAdmin extends Closeable {
   /** @return the topic name and its configurations. */
   Map<String, TopicConfig> topics();
 
+  default Set<String> publicTopicNames() {
+    return publicTopics().keySet();
+  }
+
+  /** @return the topic name and its configurations. (Without internal topics) */
+  Map<String, TopicConfig> publicTopics();
+
   /** @return a topic creator to set all topic configs and then run the procedure. */
   Creator creator();
 
diff --git a/app/src/main/java/org/astraea/topic/cost/GetPartitionInf.java b/app/src/main/java/org/astraea/topic/cost/GetPartitionInf.java
index 3a597872..4216ea8b 100644
--- a/app/src/main/java/org/astraea/topic/cost/GetPartitionInf.java
+++ b/app/src/main/java/org/astraea/topic/cost/GetPartitionInf.java
@@ -10,25 +10,25 @@ public class GetPartitionInf {
     Map<Integer, Map<TopicPartition, Integer>> brokerPartitionSize = new HashMap<>();
     for (var broker : client.brokerIds()) {
       Map<TopicPartition, Integer> partitionSize = new HashMap<>();
-      for (var partition : client.replicas(client.topics().keySet()).keySet())
-        if (client.replicas(client.topics().keySet()).get(partition).get(0).broker() == broker)
+      for (var partition : client.replicas(client.publicTopics().keySet()).keySet())
+        if (client.replicas(client.publicTopics().keySet()).get(partition).get(0).broker()
+            == broker)
           partitionSize.put(
               partition,
-              (int) client.replicas(client.topics().keySet()).get(partition).get(0).size());
+              (int) client.replicas(client.publicTopics().keySet()).get(partition).get(0).size());
       brokerPartitionSize.put(broker, partitionSize);
     }
     return brokerPartitionSize;
   }
 
   static Map<String, Integer> getRetentionMillis(TopicAdmin client) {
-    return client.topics().entrySet().stream()
+    return client.publicTopics().entrySet().stream()
         .collect(
             Collectors.toMap(
                 Map.Entry::getKey,
                 entry ->
-                    client
-                        .topics()
-                        .get(entry.getKey())
+                    entry
+                        .getValue()
                         .value("retention.ms")
                         .map(Integer::parseInt)
                         .orElseThrow(
diff --git a/app/src/main/java/org/astraea/utils/DataRate.java b/app/src/main/java/org/astraea/utils/DataRate.java
new file mode 100644
index 00000000..94bd315a
--- /dev/null
+++ b/app/src/main/java/org/astraea/utils/DataRate.java
@@ -0,0 +1,184 @@
+package org.astraea.utils;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.MathContext;
+import java.math.RoundingMode;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Comparator;
+import java.util.Locale;
+
+/** Data rate class */
+public class DataRate {
+  private final DataSize totalBitTransmitted;
+  private final BigDecimal durationInNanoSecond;
+
+  static BigDecimal fromDurationToBigDecimalSafely(Duration duration) {
+    // It the given duration is extremely long(like 1000 years), it might overflow the long
+    // maximum value in nano second unit.
+    if (duration.toSeconds() >= Long.MAX_VALUE / 1e9) {
+      return BigDecimal.valueOf(duration.toSeconds())
+          .multiply(BigDecimal.valueOf(ChronoUnit.SECONDS.getDuration().toNanos()))
+          .add(BigDecimal.valueOf(duration.toNanosPart()));
+    } else {
+      return BigDecimal.valueOf(duration.toNanos());
+    }
+  }
+
+  static BigInteger fromDurationToBigIntegerSafely(Duration duration) {
+    // It the given duration is extremely long(like 1000 years), it might overflow the long
+    // maximum value in nano second unit.
+    if (duration.toSeconds() >= Long.MAX_VALUE / 1e9) {
+      return BigInteger.valueOf(duration.toSeconds())
+          .multiply(BigInteger.valueOf(ChronoUnit.SECONDS.getDuration().toNanos()))
+          .add(BigInteger.valueOf(duration.toNanosPart()));
+    } else {
+      return BigInteger.valueOf(duration.toNanos());
+    }
+  }
+
+  DataRate(DataSize initialSize, Duration duration) {
+    this.totalBitTransmitted = initialSize;
+    this.durationInNanoSecond = fromDurationToBigDecimalSafely(duration);
+  }
+
+  /** Return the ideal data rate under specific time unit. */
+  public BigDecimal idealDataRate(ChronoUnit chronoUnit) {
+    return toBigDecimal(idealDataUnit(chronoUnit), chronoUnit);
+  }
+
+  /** Return the ideal data unit under specific time unit. */
+  public DataUnit idealDataUnit(ChronoUnit chronoUnit) {
+    final BigInteger dataRate =
+        totalBitTransmitted
+            .bits()
+            .multiply(fromDurationToBigIntegerSafely(chronoUnit.getDuration()))
+            .divide(durationInNanoSecond.toBigInteger());
+
+    return DataUnit.BYTE_UNIT_SIZE_ORDERED_LIST.stream()
+        .sorted(Comparator.reverseOrder())
+        .dropWhile((x) -> dataRate.compareTo(x.bits) < 0)
+        .findFirst()
+        .orElse(DataUnit.Byte);
+  }
+
+  /** Calculate the data rate in a specific {@link DataUnit} over a specific {@link ChronoUnit}. */
+  public BigDecimal toBigDecimal(DataUnit dataUnit, ChronoUnit chronoUnit) {
+    return toBigDecimal(dataUnit, chronoUnit.getDuration());
+  }
+
+  /** Calculate the data rate in a specific {@link DataUnit} over a specific {@link Duration}. */
+  public BigDecimal toBigDecimal(DataUnit dataUnit, Duration duration) {
+    return this.totalBitTransmitted
+        .measurement(dataUnit)
+        .multiply(fromDurationToBigDecimalSafely(duration))
+        .divide(durationInNanoSecond, MathContext.DECIMAL32);
+  }
+
+  /**
+   * Create a {@link DataRate} object from data volume and time unit
+   *
+   * <pre>{@code
+   * DataRate.of(100, DataUnit.KB, ChronoUnit.SECONDS);    // 100 KB/s
+   * DataRate.of(100, DataUnit.MB, ChronoUnit.MONTHS);     // 50 MB/month
+   * }</pre>
+   *
+   * @return an object {@link DataRate} correspond to given arguments
+   */
+  public static DataRate of(long measurement, DataUnit dataUnit, ChronoUnit chronoUnit) {
+    return of(measurement, dataUnit, chronoUnit.getDuration());
+  }
+
+  /**
+   * Create a {@link DataRate} object from data volume and time duration
+   *
+   * <pre>{@code
+   * DataRate.of(100, DataUnit.KB, Duration.ofSeconds(1)); // 100 KB/s
+   * DataRate.of(100, DataUnit.MB, Duration.ofSeconds(2)); // 50 MB/s
+   * }</pre>
+   *
+   * @return an object {@link DataRate} correspond to given arguments
+   */
+  public static DataRate of(long measurement, DataUnit dataUnit, Duration duration) {
+    return new DataRate(dataUnit.of(measurement), duration);
+  }
+
+  /**
+   * Create a {@link DataRate} object from data volume and time duration
+   *
+   * <pre>{@code
+   * DataRate.of(DataUnit.KB.of(100), ChronoUnit.SECONDS); // 100 KB/s
+   * DataRate.of(DataUnit.MB.of(100), ChronoUnit.SECONDS); // 50 MB/s
+   * }</pre>
+   *
+   * @return an object {@link DataRate} correspond to given arguments
+   */
+  public static DataRate of(DataSize dataSize, ChronoUnit chronoUnit) {
+    return new DataRate(dataSize, chronoUnit.getDuration());
+  }
+
+  /**
+   * Create a {@link DataRate} object from data volume and time duration
+   *
+   * <pre>{@code
+   * DataRate.of(DataUnit.KB.of(100), Duration.ofSeconds(1)); // 100 KB/s
+   * DataRate.of(DataUnit.MB.of(100), Duration.ofSeconds(2)); // 50 MB/s
+   * }</pre>
+   *
+   * @return an object {@link DataRate} correspond to given arguments
+   */
+  public static DataRate of(DataSize dataSize, Duration duration) {
+    return new DataRate(dataSize, duration);
+  }
+
+  public static BigDecimal ofBigDecimal(long measurement, DataUnit unit, ChronoUnit time) {
+    return ofBigDecimal(unit.of(measurement), unit, time.getDuration());
+  }
+
+  public static BigDecimal ofBigDecimal(long measurement, DataUnit unit, Duration time) {
+    return ofBigDecimal(unit.of(measurement), unit, time);
+  }
+
+  public static BigDecimal ofBigDecimal(DataSize size, DataUnit unit, ChronoUnit time) {
+    return ofBigDecimal(size, unit, time.getDuration());
+  }
+
+  public static BigDecimal ofBigDecimal(DataSize size, DataUnit unit, Duration time) {
+    return of(size, time).toBigDecimal(unit, time);
+  }
+
+  public static double ofDouble(long measurement, DataUnit unit, ChronoUnit time) {
+    return ofBigDecimal(measurement, unit, time).doubleValue();
+  }
+
+  public static double ofDouble(long measurement, DataUnit unit, Duration time) {
+    return ofBigDecimal(measurement, unit, time).doubleValue();
+  }
+
+  public static double ofDouble(DataSize size, DataUnit unit, ChronoUnit time) {
+    return ofBigDecimal(size, unit, time).doubleValue();
+  }
+
+  public static double ofDouble(DataSize size, DataUnit unit, Duration time) {
+    return ofBigDecimal(size, unit, time).doubleValue();
+  }
+
+  private static String chronoName(ChronoUnit chronoUnit) {
+    return chronoUnit.name().toLowerCase(Locale.ROOT);
+  }
+
+  public String toString(DataUnit dataUnit, ChronoUnit chronoUnit) {
+    BigDecimal dataRate = toBigDecimal(dataUnit, chronoUnit).setScale(2, RoundingMode.HALF_EVEN);
+    return String.format("%s %s/%s", dataRate, dataUnit.name(), chronoName(chronoUnit));
+  }
+
+  public String toString(ChronoUnit chronoUnit) {
+    return toString(idealDataUnit(chronoUnit), chronoUnit);
+  }
+
+  @Override
+  public String toString() {
+    return toString(ChronoUnit.SECONDS);
+  }
+}
diff --git a/app/src/main/java/org/astraea/utils/DataSize.java b/app/src/main/java/org/astraea/utils/DataSize.java
new file mode 100644
index 00000000..a30a9ab7
--- /dev/null
+++ b/app/src/main/java/org/astraea/utils/DataSize.java
@@ -0,0 +1,211 @@
+package org.astraea.utils;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.MathContext;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Comparator;
+import java.util.Objects;
+
+/** Data size class */
+public class DataSize implements Comparable<DataSize> {
+
+  private final BigInteger bits;
+
+  DataSize(long volume, DataUnit dataUnit) {
+    this(BigInteger.valueOf(volume).multiply(dataUnit.bits));
+  }
+
+  DataSize(BigInteger bigInteger) {
+    this.bits = bigInteger;
+  }
+
+  /**
+   * Add data volume.
+   *
+   * @param measurement data measurement.
+   * @param dataUnit data unit.
+   * @return a new {@link DataSize} that have applied the math operation.
+   */
+  public DataSize add(long measurement, DataUnit dataUnit) {
+    return add(new DataSize(measurement, Objects.requireNonNull(dataUnit)));
+  }
+
+  /**
+   * Subtract data volume.
+   *
+   * @param measurement data measurement.
+   * @param dataUnit data unit.
+   * @return a new {@link DataSize} that have applied the math operation.
+   */
+  public DataSize subtract(long measurement, DataUnit dataUnit) {
+    return subtract(new DataSize(measurement, Objects.requireNonNull(dataUnit)));
+  }
+
+  /**
+   * Multiply current data volume by a scalar.
+   *
+   * @param scalar a long integer.
+   * @return a new {@link DataSize} that have applied the math operation.
+   */
+  public DataSize multiply(long scalar) {
+    return multiply(BigInteger.valueOf(scalar));
+  }
+
+  /**
+   * Divide current data volume by a scalar.
+   *
+   * @param scalar a long integer.
+   * @return a new {@link DataSize} that have applied the math operation.
+   */
+  public DataSize divide(long scalar) {
+    return divide(BigInteger.valueOf(scalar));
+  }
+
+  /**
+   * Add by given {@link DataSize}.
+   *
+   * @param rhs the right hand side value.
+   * @return a new {@link DataSize} that have applied the math operation.
+   */
+  public DataSize add(DataSize rhs) {
+    return add(Objects.requireNonNull(rhs).bits);
+  }
+
+  /**
+   * Subtract by given {@link DataSize}.
+   *
+   * @param rhs the right hand size value.
+   * @return a new {@link DataSize} that have applied the math operation.
+   */
+  public DataSize subtract(DataSize rhs) {
+    return subtract(Objects.requireNonNull(rhs).bits);
+  }
+
+  public boolean smallerThan(DataSize dataSize) {
+    return this.compareTo(dataSize) < 0;
+  }
+
+  public boolean smallerEqualTo(DataSize dataSize) {
+    return this.compareTo(dataSize) <= 0;
+  }
+
+  public boolean greaterThan(DataSize dataSize) {
+    return this.compareTo(dataSize) > 0;
+  }
+
+  public boolean greaterEqualTo(DataSize dataSize) {
+    return this.compareTo(dataSize) >= 0;
+  }
+
+  private DataSize add(BigInteger rhs) {
+    return new DataSize(bits.add(rhs));
+  }
+
+  private DataSize subtract(BigInteger rhs) {
+    return new DataSize(bits.subtract(rhs));
+  }
+
+  private DataSize multiply(BigInteger rhs) {
+    return new DataSize(bits.multiply(rhs));
+  }
+
+  private DataSize divide(BigInteger rhs) {
+    return new DataSize(bits.divide(rhs));
+  }
+
+  /** Current bits. */
+  public BigInteger bits() {
+    return bits;
+  }
+
+  /**
+   * The measurement value in term of specific data unit.
+   *
+   * @param dataUnit data unit to describe current size.
+   * @return a {@link BigDecimal} describe current data size in term of specific data unit.
+   */
+  public BigDecimal measurement(DataUnit dataUnit) {
+    return new BigDecimal(this.bits).divide(new BigDecimal(dataUnit.bits), MathContext.DECIMAL32);
+  }
+
+  /**
+   * The measurement value in term of the most ideal data unit for human readability.
+   *
+   * @return a {@link BigDecimal} describe current data size in term of the ideal data unit.
+   */
+  public BigDecimal idealMeasurement() {
+    return measurement(idealDataUnit());
+  }
+
+  /**
+   * Return the most ideal data unit in terms of human readability. Usually, the most suitable unit
+   * describes the data volume in the fewest decimal digits.
+   *
+   * <p>For example, given 1,000,000,000 bytes. We can describe this data volume in various way:
+   *
+   * <ol>
+   *   <li>8,000,000,000 Bit
+   *   <li>1,000,000,000 Byte
+   *   <li>1,000,000 KB
+   *   <li>1,000 MB
+   *   <li>1 GB
+   *   <li>0.0001 TB
+   * </ol>
+   *
+   * <p>For all these possibilities, GB is more ideal since it has the fewest decimal digits.
+   *
+   * @return a {@link DataUnit} that is suitable to describe current data volume.
+   */
+  public DataUnit idealDataUnit() {
+    return DataUnit.BYTE_UNIT_SIZE_ORDERED_LIST.stream()
+        .sorted(Comparator.reverseOrder())
+        .dropWhile((x) -> this.bits.compareTo(x.bits) < 0)
+        .findFirst()
+        .orElse(DataUnit.Byte);
+  }
+
+  /** Return a {@link DataRate} based on current data size over a specific time unit. */
+  public DataRate dataRate(ChronoUnit chronoUnit) {
+    return DataRate.of(this, chronoUnit.getDuration());
+  }
+
+  /**
+   * Return a {@link DataRate} based on current data size over a specific {@link Duration} of time.
+   */
+  public DataRate dataRate(Duration timePassed) {
+    return DataRate.of(this, timePassed);
+  }
+
+  /** Return a string represent current size in given data unit. */
+  public String toString(DataUnit unit) {
+    return String.format("%s %s", this.bits.divide(unit.bits), unit.name());
+  }
+
+  /**
+   * Return a string representing the current size, the string will use an easy to read data unit.
+   */
+  @Override
+  public String toString() {
+    return toString(idealDataUnit());
+  }
+
+  @Override
+  public int compareTo(DataSize hs) {
+    return this.bits.compareTo(hs.bits);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    DataSize size = (DataSize) o;
+    return bits.equals(size.bits);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(bits);
+  }
+}
diff --git a/app/src/main/java/org/astraea/utils/DataUnit.java b/app/src/main/java/org/astraea/utils/DataUnit.java
new file mode 100644
index 00000000..f6586081
--- /dev/null
+++ b/app/src/main/java/org/astraea/utils/DataUnit.java
@@ -0,0 +1,121 @@
+package org.astraea.utils;
+
+import java.math.BigInteger;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * Utility related to data and unit, this class dedicated to easing the pain of converting and
+ * calculating/displaying/measurement and its unit.
+ *
+ * @see <a href="https://physics.nist.gov/cuu/Units/binary.html">NIST Reference on Binary Unit</a>
+ *     to understand the detail behind those units.
+ */
+public enum DataUnit {
+  Bit(1, false),
+  Kb(1000, Bit),
+  Mb(1000, Kb),
+  Gb(1000, Mb),
+  Tb(1000, Gb),
+  Pb(1000, Tb),
+  Eb(1000, Pb),
+  Zb(1000, Eb),
+  Yb(1000, Zb),
+
+  Kib(1024, Bit),
+  Mib(1024, Kib),
+  Gib(1024, Mib),
+  Tib(1024, Gib),
+  Pib(1024, Tib),
+  Eib(1024, Pib),
+  Zib(1024, Eib),
+  Yib(1024, Zib),
+
+  Byte(8, true, true),
+  KB(1000, Byte, true),
+  MB(1000, KB, true),
+  GB(1000, MB, true),
+  TB(1000, GB, true),
+  PB(1000, TB, true),
+  EB(1000, PB, true),
+  ZB(1000, EB, true),
+  YB(1000, ZB, true),
+
+  KiB(1024, Byte),
+  MiB(1024, KiB),
+  GiB(1024, MiB),
+  TiB(1024, GiB),
+  PiB(1024, TiB),
+  EiB(1024, PiB),
+  ZiB(1024, EiB),
+  YiB(1024, ZiB);
+
+  final BigInteger bits;
+  final boolean byteBasedUnit;
+  final boolean candidateUnitForToString;
+
+  DataUnit(long bits, boolean isByteBasedUnit) {
+    this(bits, isByteBasedUnit, false);
+  }
+
+  DataUnit(long bits, boolean isByteBasedUnit, boolean candidateUnitForToString) {
+    this.bits = BigInteger.valueOf(bits);
+    this.byteBasedUnit = isByteBasedUnit;
+    this.candidateUnitForToString = candidateUnitForToString;
+  }
+
+  DataUnit(long measurement, DataUnit unit) {
+    this(measurement, unit, false);
+  }
+
+  DataUnit(long measurement, DataUnit unit, boolean candidateUnitForToString) {
+    this.bits = unit.bits.multiply(BigInteger.valueOf(measurement));
+    this.byteBasedUnit = unit.byteBasedUnit;
+    this.candidateUnitForToString = candidateUnitForToString;
+  }
+
+  /**
+   * Return a {@link DataSize} based on given measurement and related unit.
+   *
+   * <pre>{@code
+   * DataUnit.KB.of(500);  // 500 KB  (500 * 1000 bytes)
+   * DataUnit.KiB.of(500); // 500 KiB (500 * 1024 bytes)
+   * DataUnit.Kb.of(500);  // 500 Kb  (500 * 1000 bits)
+   * DataUnit.Kib.of(500); // 500 Kib (500 * 1024 bits)
+   * }</pre>
+   *
+   * @param measurement the data size measurement.
+   * @return a size object of given measurement under specific data unit.
+   */
+  public DataSize of(long measurement) {
+    return new DataSize(measurement, this);
+  }
+
+  /**
+   * Return a {@link DataSize} based on given measurement and unit.
+   *
+   * <pre>{@code
+   * DataUnit.of(500, DataUnit.KB);   // 500 KB  (500 * 1000 bytes)
+   * DataUnit.of(500, DataUnit.KiB);  // 500 KiB (500 * 1024 bytes)
+   * DataUnit.of(500, DataUnit.Kb);   // 500 Kb  (500 * 1000 bits)
+   * DataUnit.of(500, DataUnit.Kib);  // 500 Kib (500 * 1024 bits)
+   * }</pre>
+   *
+   * @param measurement the data size measurement.
+   * @param unit the data unit of given measurement.
+   * @return a size object of given measurement under specific data unit.
+   */
+  public static DataSize of(long measurement, DataUnit unit) {
+    return new DataSize(measurement, unit);
+  }
+
+  /**
+   * List of recommended unit for display data, {@link DataSize#toString()} take advantage of this
+   * collection to find ideal unit for string formalization.
+   */
+  static final List<DataUnit> BYTE_UNIT_SIZE_ORDERED_LIST =
+      Arrays.stream(DataUnit.values())
+          .filter(x -> x.candidateUnitForToString)
+          .sorted(Comparator.comparing(x -> x.bits))
+          .collect(Collectors.toUnmodifiableList());
+}
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index e750102e..d2880ba8 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.3.2-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
similarity index 80%
rename from app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java
rename to app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
index b3159a3a..7526f7e7 100644
--- a/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java
+++ b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
@@ -3,6 +3,7 @@ package org.astraea.metrics.collector;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
@@ -15,7 +16,7 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
-public class TestBeanCollector {
+public class BeanCollectorTest {
   private final HasBeanObject mbean = Mockito.mock(HasBeanObject.class);
   private final MBeanClient mbeanClient = Mockito.mock(MBeanClient.class);
   private final BiFunction<String, Integer, MBeanClient> clientCreator =
@@ -142,13 +143,13 @@ public class TestBeanCollector {
     var receivers = receivers(collector);
     receivers.forEach(Receiver::current);
 
-    Assertions.assertEquals(1, collector.clients.size());
+    Assertions.assertEquals(1, collector.nodes.size());
     Assertions.assertEquals(
-        mbeanClient, collector.clients.entrySet().iterator().next().getValue().mBeanClient);
+        mbeanClient, collector.nodes.entrySet().iterator().next().getValue().mBeanClient);
 
     receivers.forEach(Receiver::close);
-    Assertions.assertEquals(1, collector.clients.size());
-    Assertions.assertNull(collector.clients.entrySet().iterator().next().getValue().mBeanClient);
+    Assertions.assertEquals(1, collector.nodes.size());
+    Assertions.assertNull(collector.nodes.entrySet().iterator().next().getValue().mBeanClient);
   }
 
   @Test
@@ -205,4 +206,40 @@ public class TestBeanCollector {
       sleep(1);
     }
   }
+
+  @Test
+  void testMultiplesReceiversData() {
+    var collector =
+        BeanCollector.builder()
+            .interval(Duration.ofSeconds(1))
+            .clientCreator(clientCreator)
+            .build();
+
+    var receivers =
+        IntStream.range(0, 2)
+            .mapToObj(
+                i -> {
+                  var count = new AtomicInteger();
+                  return Map.entry(
+                      count,
+                      collector
+                          .register()
+                          .host("unknown")
+                          .port(100)
+                          .metricsGetter(
+                              c -> {
+                                count.incrementAndGet();
+                                return mbean;
+                              })
+                          .build());
+                })
+            .collect(Collectors.toList());
+
+    sleep(1);
+    receivers.forEach(e -> Assertions.assertEquals(0, e.getKey().get()));
+
+    sleep(1);
+    receivers.forEach(e -> Assertions.assertEquals(1, e.getValue().current().size()));
+    receivers.forEach(e -> Assertions.assertEquals(1, e.getKey().get()));
+  }
 }
diff --git a/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorTest.java b/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorTest.java
index 27542d65..c238a757 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorTest.java
@@ -6,6 +6,7 @@ import static org.mockito.Mockito.*;
 import java.io.ByteArrayOutputStream;
 import java.io.PrintStream;
 import java.time.Duration;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -17,6 +18,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.kafka.common.TopicPartition;
 import org.astraea.argument.ArgumentUtil;
+import org.astraea.utils.DataUnit;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -311,13 +313,16 @@ class ReplicaSyncingMonitorTest {
       double expectedProgress,
       int expectedRemainingTime) {
     // act
-    var dataRate =
-        new ReplicaSyncingMonitor.DataRate(
-            leaderSize, previousSize, currentSize, Duration.ofMillis(interval));
+    var progress =
+        new ReplicaSyncingMonitor.ProgressInfo(
+            DataUnit.Byte.of(leaderSize),
+            DataUnit.Byte.of(previousSize),
+            DataUnit.Byte.of(currentSize),
+            Duration.ofMillis(interval));
 
     // assert
-    assertEquals(expectedDataRatePerSec, dataRate.dataRatePerSec());
-    assertEquals(expectedProgress, dataRate.progress());
-    assertEquals(Duration.ofSeconds(expectedRemainingTime), dataRate.estimateFinishTime());
+    assertEquals(expectedDataRatePerSec, progress.dataRate(DataUnit.Byte, ChronoUnit.SECONDS));
+    assertEquals(expectedProgress, progress.progress());
+    assertEquals(Duration.ofSeconds(expectedRemainingTime), progress.estimateFinishTime());
   }
 }
diff --git a/app/src/test/java/org/astraea/utils/DataRateTest.java b/app/src/test/java/org/astraea/utils/DataRateTest.java
new file mode 100644
index 00000000..ff4b9967
--- /dev/null
+++ b/app/src/test/java/org/astraea/utils/DataRateTest.java
@@ -0,0 +1,84 @@
+package org.astraea.utils;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+class DataRateTest {
+
+  static void assertFloatingValueEquals(double expected, double actual) {
+    assertTrue(
+        Math.abs(expected - actual) < 1e-9,
+        () ->
+            String.format(
+                "floating value might not equal. Expected: \"%f\" but Actual: \"%f\" and difference is \"%f\"",
+                expected, actual, Math.abs(expected - actual)));
+  }
+
+  @Test
+  void fromDurationSafely() {
+    assertDoesNotThrow(
+        () -> {
+          DataRate.fromDurationToBigDecimalSafely(ChronoUnit.FOREVER.getDuration());
+        });
+    assertDoesNotThrow(
+        () -> {
+          DataRate.fromDurationToBigIntegerSafely(ChronoUnit.FOREVER.getDuration());
+        });
+  }
+
+  @Test
+  void idealDataRateAndUnit() {
+    var sut = DataRate.of(1000, DataUnit.KB, Duration.ofSeconds(1));
+
+    assertFloatingValueEquals(1.0, sut.idealDataRate(ChronoUnit.SECONDS).doubleValue());
+    assertSame(DataUnit.MB, sut.idealDataUnit(ChronoUnit.SECONDS));
+  }
+
+  @Test
+  void dataRate() {
+    var sut0 =
+        DataRate.of(500, DataUnit.KB, Duration.ofSeconds(1))
+            .toBigDecimal(DataUnit.MB, Duration.ofSeconds(1));
+    var sut1 =
+        DataRate.of(500, DataUnit.KB, ChronoUnit.SECONDS)
+            .toBigDecimal(DataUnit.MB, Duration.ofSeconds(1));
+
+    assertFloatingValueEquals(0.5, sut0.doubleValue());
+    assertFloatingValueEquals(0.5, sut1.doubleValue());
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      value = {
+        // measurement, dataUnit, passedSecond, expectedIdealDataRate, expectedDataUnit
+        "            1, Bit     , 1           ,                 0.125, Byte            ",
+        "            1, YB      , 1000000000  ,                     1, PB              ",
+        "           16, Bit     , 2           ,                     1, Byte            ",
+        "          100, KB      , 1           ,                   100, KB              ",
+        "          999, MB      , 1           ,                   999, MB              ",
+        "         1000, KB      , 1           ,                     1, MB              ",
+        "         1000, KB      , 1           ,                     1, MB              ",
+        "         1234, MB      , 1           ,                 1.234, GB              ",
+        "         1234, MB      , 1           ,                 1.234, GB              ",
+        "         5000, KiB     , 1           ,                  5.12, MB              ",
+        "      1000000, Bit     , 8           ,                15.625, KB              ",
+        "      1000000, MB      , 1           ,                     1, TB              "
+      })
+  void testDataRate(
+      long measurement,
+      DataUnit dataUnit,
+      long passedSecond,
+      double expectedIdealDataRate,
+      DataUnit expectedDataUnit) {
+    DataRate sut = DataRate.of(measurement, dataUnit, Duration.ofSeconds(passedSecond));
+
+    assertFloatingValueEquals(
+        expectedIdealDataRate, sut.idealDataRate(ChronoUnit.SECONDS).doubleValue());
+    assertEquals(expectedDataUnit, sut.idealDataUnit(ChronoUnit.SECONDS));
+  }
+}
diff --git a/app/src/test/java/org/astraea/utils/DataSizeTest.java b/app/src/test/java/org/astraea/utils/DataSizeTest.java
new file mode 100644
index 00000000..83d1ca6d
--- /dev/null
+++ b/app/src/test/java/org/astraea/utils/DataSizeTest.java
@@ -0,0 +1,249 @@
+package org.astraea.utils;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+class DataSizeTest {
+
+  @SuppressWarnings("ResultOfMethodCallIgnored")
+  @Test
+  void typicalUsageOfDataUnit() {
+    // 500KB
+    DataUnit.KB.of(500);
+    // 500MB + 500MB
+    DataUnit.MB.of(500).add(500, DataUnit.MB);
+    // 500GB - 500GB
+    DataUnit.GB.of(500).subtract(500, DataUnit.GB);
+    // 500 TB / 10
+    DataUnit.TB.of(500).divide(10);
+    // 500 PB * 10
+    DataUnit.PB.of(500).multiply(10);
+
+    // the data rate of sending 1 EB over 1 second
+    DataUnit.EB.of(1).dataRate(ChronoUnit.SECONDS);
+    // the data rate of sending 1 EB over 1 millisecond
+    DataUnit.EB.of(1).dataRate(ChronoUnit.MILLIS);
+    // the data rate of sending 1 EB over 1 century (100 years)
+    DataUnit.EB.of(1).dataRate(ChronoUnit.CENTURIES);
+    // the data rate of sending 1 EB over 59 seconds
+    DataUnit.EB.of(1).dataRate(Duration.ofSeconds(59));
+    // data rate string of sending 1 EB over 1 second in the most human friendly data unit & second
+    // time unit
+    DataUnit.EB.of(1).dataRate(ChronoUnit.SECONDS).toString();
+    // data rate string of sending 1 EB over 1 second in the most human friendly data unit & 1 hour
+    // time unit
+    DataUnit.EB.of(1).dataRate(ChronoUnit.SECONDS).toString(ChronoUnit.HOURS);
+    // data rate string of sending 1 EB over 1 second in ZB unit and hour time unit
+    DataUnit.EB.of(1).dataRate(ChronoUnit.SECONDS).toString(DataUnit.ZB, ChronoUnit.HOURS);
+
+    // someone wondering that if we send 1 YB of data over 1000 years, how much data we sent per
+    // second.
+    DataUnit.YB.of(1).dataRate(ChronoUnit.MILLENNIA).toString(ChronoUnit.SECONDS);
+
+    // faster convert between DataRate and others.
+    var randomSize = DataUnit.Byte.of(ThreadLocalRandom.current().nextLong());
+
+    BigDecimal bigDecimal0 = DataRate.ofBigDecimal(1000, DataUnit.Byte, ChronoUnit.SECONDS);
+    BigDecimal bigDecimal1 = DataRate.ofBigDecimal(1000, DataUnit.Byte, Duration.ofSeconds(1));
+    BigDecimal bigDecimal2 = DataRate.ofBigDecimal(randomSize, DataUnit.Byte, ChronoUnit.SECONDS);
+    BigDecimal bigDecimal3 =
+        DataRate.ofBigDecimal(randomSize, DataUnit.Byte, Duration.ofSeconds(1));
+
+    double double0 = DataRate.ofDouble(1000, DataUnit.Byte, ChronoUnit.SECONDS);
+    double double1 = DataRate.ofDouble(1000, DataUnit.Byte, Duration.ofSeconds(1));
+    double double2 = DataRate.ofDouble(randomSize, DataUnit.Byte, ChronoUnit.SECONDS);
+    double double3 = DataRate.ofDouble(randomSize, DataUnit.Byte, Duration.ofSeconds(1));
+
+    // solve the above problem
+    var dataVolume = DataUnit.YB.of(1);
+    var dataVolumeOver1000Years = dataVolume.dataRate(ChronoUnit.MILLENNIA);
+    Consumer<ChronoUnit> tellMeTheAnswerIn =
+        (ChronoUnit chronoUnit) ->
+            System.out.printf(
+                "If Bob sends %s %s data over 1000 years. Then Bob has to send %s %s each %s.%n",
+                dataVolume.idealMeasurement(),
+                dataVolume.idealDataUnit(),
+                dataVolumeOver1000Years.toBigDecimal(
+                    dataVolumeOver1000Years.idealDataUnit(chronoUnit), chronoUnit),
+                dataVolumeOver1000Years.idealDataUnit(chronoUnit),
+                chronoUnit);
+
+    tellMeTheAnswerIn.accept(ChronoUnit.NANOS);
+    tellMeTheAnswerIn.accept(ChronoUnit.SECONDS);
+    tellMeTheAnswerIn.accept(ChronoUnit.HOURS);
+    tellMeTheAnswerIn.accept(ChronoUnit.DAYS);
+    tellMeTheAnswerIn.accept(ChronoUnit.MONTHS);
+    tellMeTheAnswerIn.accept(ChronoUnit.YEARS);
+    tellMeTheAnswerIn.accept(ChronoUnit.DECADES);
+    tellMeTheAnswerIn.accept(ChronoUnit.CENTURIES);
+    tellMeTheAnswerIn.accept(ChronoUnit.MILLENNIA);
+
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.NANOS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.MICROS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.MILLIS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.SECONDS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.MINUTES));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.HOURS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.DAYS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.WEEKS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.MONTHS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.YEARS));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.DECADES));
+    System.out.println(dataVolumeOver1000Years.toString(ChronoUnit.CENTURIES));
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiterString = ",",
+      value = {
+        // Unit Name, bits for 1 unit
+        "  Bit      , 1",
+        "  Kb       , 1000",
+        "  Mb       , 1000000",
+        "  Gb       , 1000000000",
+        "  Tb       , 1000000000000",
+        "  Pb       , 1000000000000000",
+        "  Eb       , 1000000000000000000",
+        "  Zb       , 1000000000000000000000",
+        "  Yb       , 1000000000000000000000000",
+        "  Kib      , 1024",
+        "  Mib      , 1048576",
+        "  Gib      , 1073741824",
+        "  Tib      , 1099511627776",
+        "  Pib      , 1125899906842624",
+        "  Eib      , 1152921504606846976",
+        "  Zib      , 1180591620717411303424",
+        "  Yib      , 1208925819614629174706176",
+        "  Byte     , 8",
+        "  KB       , 8000",
+        "  MB       , 8000000",
+        "  GB       , 8000000000",
+        "  TB       , 8000000000000",
+        "  PB       , 8000000000000000",
+        "  EB       , 8000000000000000000",
+        "  ZB       , 8000000000000000000000",
+        "  YB       , 8000000000000000000000000",
+        "  KiB      , 8192",
+        "  MiB      , 8388608",
+        "  GiB      , 8589934592",
+        "  TiB      , 8796093022208",
+        "  PiB      , 9007199254740992",
+        "  EiB      , 9223372036854775808",
+        "  ZiB      , 9444732965739290427392",
+        "  YiB      , 9671406556917033397649408",
+      })
+  void of(String unitName, String expectedBits) {
+    assertEquals(new BigInteger(expectedBits), DataUnit.valueOf(unitName).of(1).bits());
+  }
+
+  @Test
+  void measurement() {
+    var value = DataUnit.PB.of(1);
+
+    var assertEquals =
+        (BiFunction<BigDecimal, BigDecimal, Void>)
+            (a, b) -> {
+              double v0 = a.doubleValue();
+              double v1 = b.doubleValue();
+              assertTrue(Math.abs(v0 - v1) <= 0.00000000001);
+              return null;
+            };
+
+    assertEquals.apply(new BigDecimal("1000000000"), value.measurement(DataUnit.MB));
+    assertEquals.apply(new BigDecimal("1000000"), value.measurement(DataUnit.GB));
+    assertEquals.apply(new BigDecimal("1000"), value.measurement(DataUnit.TB));
+    assertEquals.apply(new BigDecimal("1"), value.measurement(DataUnit.PB));
+    assertEquals.apply(new BigDecimal("0.001"), value.measurement(DataUnit.EB));
+    assertEquals.apply(new BigDecimal("0.000001"), value.measurement(DataUnit.ZB));
+    assertEquals.apply(new BigDecimal("0.000000001"), value.measurement(DataUnit.YB));
+  }
+
+  @Test
+  void add() {
+    var lhs = DataUnit.KB.of(1024);
+    var rhs = DataUnit.GB.of(1);
+    assertEquals(
+        1024 * 1000 + 1000 * 1000 * 1000L, lhs.add(rhs).measurement(DataUnit.Byte).longValue());
+  }
+
+  @Test
+  void subtract() {
+    var lhs = DataUnit.GB.of(1);
+    var rhs = DataUnit.KB.of(1024);
+    assertEquals(
+        1000 * 1000 * 1000 - 1024 * 1000L,
+        lhs.subtract(rhs).measurement(DataUnit.Byte).longValue());
+  }
+
+  @Test
+  void multiply() {
+    var lhs = DataUnit.GB.of(1);
+    var rhs = 100;
+    assertEquals(
+        100 * 1000 * 1000 * 1000L, lhs.multiply(rhs).measurement(DataUnit.Byte).longValue());
+  }
+
+  @Test
+  void divide() {
+    var lhs = DataUnit.GB.of(1);
+    var rhs = 100;
+    assertEquals(1000 * 1000 * 1000 / 100L, lhs.divide(rhs).measurement(DataUnit.Byte).longValue());
+  }
+
+  @Test
+  void idealUnit() {
+    var value =
+        DataUnit.Byte.of(1)
+            .multiply(1000) // 1 KB
+            .multiply(1000) // 1 MB
+            .multiply(1000) // 1 GB
+            .multiply(1000) // 1 TB
+            .multiply(1000) // 1 PB
+            .multiply(1000) // 1 EB
+            .multiply(1000) // 1 ZB
+            .multiply(1000) // 1 YB
+            .subtract(1, DataUnit.Bit); // 1 YB - 1 bit
+
+    assertSame(DataUnit.ZB, value.idealDataUnit());
+  }
+
+  @Test
+  void idealMeasurement() {
+    var value = DataUnit.Byte.of(1).multiply(1000).multiply(1000).multiply(1000);
+
+    assertSame(DataUnit.GB, value.idealDataUnit());
+    assertEquals(BigDecimal.ONE, value.idealMeasurement());
+  }
+
+  @Test
+  void bits() {
+    BigInteger bit5566 = DataUnit.of(5566, DataUnit.Bit).bits();
+
+    assertEquals(BigInteger.valueOf(5566), bit5566);
+  }
+
+  @Test
+  void compare() {
+    assertEquals(DataUnit.KB.of(1000), DataUnit.MB.of(1));
+    assertNotEquals(DataUnit.KB.of(1000), DataUnit.KiB.of(1000));
+
+    assertTrue(DataUnit.KB.of(999).smallerEqualTo(DataUnit.MB.of(1)));
+    assertTrue(DataUnit.KB.of(1000).smallerEqualTo(DataUnit.MB.of(1)));
+    assertTrue(DataUnit.KB.of(1000).greaterEqualTo(DataUnit.MB.of(1)));
+    assertTrue(DataUnit.KB.of(1001).greaterThan(DataUnit.MB.of(1)));
+    assertFalse(DataUnit.MB.of(1).smallerEqualTo(DataUnit.KB.of(999)));
+    assertFalse(DataUnit.MB.of(2).smallerEqualTo(DataUnit.KB.of(1000)));
+    assertFalse(DataUnit.MB.of(0).greaterEqualTo(DataUnit.KB.of(1000)));
+    assertFalse(DataUnit.MB.of(1).greaterThan(DataUnit.KB.of(1001)));
+  }
+}