Enable to get replica size

app/src/main/java/org/astraea/producer/Sender.java

@@ -11,11 +11,23 @@ public interface Sender<Key, Value> {
 
   Sender<Key, Value> topic(String topic);
 
+  /**
+   * define the data route if you don't want to partitioner to decide the target.
+   *
+   * @param partition target partition
+   * @return this sender
+   */
   Sender<Key, Value> partition(int partition);
 
   Sender<Key, Value> timestamp(long timestamp);
 
   Sender<Key, Value> headers(Collection<Header> headers);
 
+  /**
+   * send data to servers. This operation is running in background. You have to call {@link
+   * CompletionStage#toCompletableFuture()} to wait response of servers.
+   *
+   * @return an async operation stage.
+   */
   CompletionStage<Metadata> run();
 }

app/src/main/java/org/astraea/topic/TopicAdmin.java

@@ -290,6 +290,9 @@ public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
                                                           replicaInfo == null
                                                               ? -1
                                                               : replicaInfo.offsetLag(),
+                                                          replicaInfo == null
+                                                              ? -1
+                                                              : replicaInfo.size(),
                                                           topicPartitionInfo.leader().id()
                                                               == node.id(),
                                                           topicPartitionInfo.isr().contains(node),
@@ -441,28 +444,57 @@ public String toString() {
   class Replica {
     public final int broker;
     public final long lag;
+    public final long size;
     public final boolean leader;
     public final boolean inSync;
     public final boolean isFuture;
     public final String path;
 
     public Replica(
-        int broker, long lag, boolean leader, boolean inSync, boolean isFuture, String path) {
+        int broker,
+        long lag,
+        long size,
+        boolean leader,
+        boolean inSync,
+        boolean isFuture,
+        String path) {
       this.broker = broker;
       this.lag = lag;
+      this.size = size;
       this.leader = leader;
       this.inSync = inSync;
       this.isFuture = isFuture;
       this.path = path;
     }
 
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      Replica replica = (Replica) o;
+      return broker == replica.broker
+          && lag == replica.lag
+          && size == replica.size
+          && leader == replica.leader
+          && inSync == replica.inSync
+          && isFuture == replica.isFuture
+          && path.equals(replica.path);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(broker, lag, size, leader, inSync, isFuture, path);
+    }
+
     @Override
     public String toString() {
       return "Replica{"
           + "broker="
           + broker
           + ", lag="
           + lag
+          + ", size="
+          + size
           + ", leader="
           + leader
           + ", inSync="
@@ -474,23 +506,5 @@ public String toString() {
           + '\''
           + '}';
     }
-
-    @Override
-    public boolean equals(Object o) {
-      if (this == o) return true;
-      if (o == null || getClass() != o.getClass()) return false;
-      Replica replica = (Replica) o;
-      return broker == replica.broker
-          && lag == replica.lag
-          && leader == replica.leader
-          && inSync == replica.inSync
-          && isFuture == replica.isFuture
-          && Objects.equals(path, replica.path);
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hash(broker, lag, leader, inSync, isFuture, path);
-    }
   }
 }

app/src/test/java/org/astraea/topic/TopicAdminTest.java

@@ -3,9 +3,11 @@
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.astraea.Utils;
+import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -86,4 +88,36 @@ void testReassign() throws IOException, InterruptedException {
           });
     }
   }
+
+  @Test
+  void testReplicaSize() throws IOException, ExecutionException, InterruptedException {
+    var topicName = "testReplicaSize";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers());
+        var producer = Producer.builder().brokers(bootstrapServers()).build()) {
+      producer.sender().topic(topicName).key(new byte[100]).run().toCompletableFuture().get();
+      var originSize =
+          topicAdmin
+              .replicas(Set.of(topicName))
+              .entrySet()
+              .iterator()
+              .next()
+              .getValue()
+              .get(0)
+              .size;
+
+      // add data again
+      producer.sender().topic(topicName).key(new byte[100]).run().toCompletableFuture().get();
+
+      var newSize =
+          topicAdmin
+              .replicas(Set.of(topicName))
+              .entrySet()
+              .iterator()
+              .next()
+              .getValue()
+              .get(0)
+              .size;
+      Assertions.assertTrue(newSize > originSize);
+    }
+  }
 }