Repository: skiptests/astraea
Pull request #344 (2022-05-15 16:33:33)

Enable to fetch producer state by Restful APIs
新增Restful APIs可以調閱producer state


Patch:
diff --git a/app/src/main/java/org/astraea/admin/Builder.java b/app/src/main/java/org/astraea/admin/Builder.java
index 4b2921c0..05700d54 100644
--- a/app/src/main/java/org/astraea/admin/Builder.java
+++ b/app/src/main/java/org/astraea/admin/Builder.java
@@ -99,6 +99,7 @@ public class Builder {
                   .get()
                   .entrySet()
                   .stream()
+                  .filter(e -> !e.getValue().activeProducers().isEmpty())
                   .collect(
                       Collectors.toMap(
                           e -> TopicPartition.from(e.getKey()),
diff --git a/app/src/main/java/org/astraea/admin/TopicPartition.java b/app/src/main/java/org/astraea/admin/TopicPartition.java
index 5d09381c..ecb45d1e 100644
--- a/app/src/main/java/org/astraea/admin/TopicPartition.java
+++ b/app/src/main/java/org/astraea/admin/TopicPartition.java
@@ -12,6 +12,23 @@ public final class TopicPartition implements Comparable<TopicPartition> {
     return new org.apache.kafka.common.TopicPartition(tp.topic(), tp.partition());
   }
 
+  public static TopicPartition of(String topic, String partition) {
+    return of(topic + "-" + partition);
+  }
+
+  public static TopicPartition of(String value) {
+    var lhs = value.lastIndexOf("-");
+    if (lhs <= 0 || lhs == value.length() - 1)
+      throw new IllegalArgumentException(
+          value + " has illegal format. It should be {topic}-{partition}");
+    try {
+      return new TopicPartition(
+          value.substring(0, lhs), Integer.parseInt(value.substring(lhs + 1)));
+    } catch (NumberFormatException e) {
+      throw new IllegalArgumentException("partition id must be number");
+    }
+  }
+
   private final int partition;
   private final String topic;
 
diff --git a/app/src/main/java/org/astraea/web/Handler.java b/app/src/main/java/org/astraea/web/Handler.java
index 2f3f280e..d9efb078 100644
--- a/app/src/main/java/org/astraea/web/Handler.java
+++ b/app/src/main/java/org/astraea/web/Handler.java
@@ -17,8 +17,7 @@ interface Handler extends HttpHandler {
 
   static <T> Set<T> compare(Set<T> all, Optional<T> target) {
     var nonexistent = target.stream().filter(id -> !all.contains(id)).collect(Collectors.toSet());
-    if (!nonexistent.isEmpty())
-      throw new NoSuchElementException("[" + nonexistent + "] does not exist");
+    if (!nonexistent.isEmpty()) throw new NoSuchElementException(nonexistent + " does not exist");
     return target.map(Set::of).orElse(all);
   }
 
diff --git a/app/src/main/java/org/astraea/web/ProducerHandler.java b/app/src/main/java/org/astraea/web/ProducerHandler.java
new file mode 100644
index 00000000..7228773e
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/ProducerHandler.java
@@ -0,0 +1,89 @@
+package org.astraea.web;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.astraea.admin.Admin;
+import org.astraea.admin.TopicPartition;
+
+class ProducerHandler implements Handler {
+
+  static final String TOPIC_KEY = "topic";
+  static final String PARTITION_KEY = "partition";
+
+  private final Admin admin;
+
+  ProducerHandler(Admin admin) {
+    this.admin = admin;
+  }
+
+  Set<TopicPartition> partitions(Map<String, String> queries) {
+    if (queries.containsKey(TOPIC_KEY) && queries.containsKey(PARTITION_KEY))
+      return Set.of(TopicPartition.of(queries.get(TOPIC_KEY), queries.get(PARTITION_KEY)));
+    var partitions = admin.partitions();
+    if (queries.containsKey(TOPIC_KEY))
+      return partitions.stream()
+          .filter(p -> p.topic().equals(queries.get(TOPIC_KEY)))
+          .collect(Collectors.toSet());
+    return partitions;
+  }
+
+  @Override
+  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+    var topics =
+        admin.producerStates(partitions(queries)).entrySet().stream()
+            .map(
+                e ->
+                    new Partition(
+                        e.getKey().topic(),
+                        e.getKey().partition(),
+                        e.getValue().stream()
+                            .map(
+                                s ->
+                                    new ProducerState(
+                                        s.producerId(),
+                                        s.producerEpoch(),
+                                        s.lastSequence(),
+                                        s.lastTimestamp()))
+                            .collect(Collectors.toUnmodifiableList())))
+            .collect(Collectors.toUnmodifiableList());
+    return new Partitions(topics);
+  }
+
+  static class ProducerState implements JsonObject {
+
+    final long producerId;
+    final int producerEpoch;
+    final int lastSequence;
+    final long lastTimestamp;
+
+    ProducerState(long producerId, int producerEpoch, int lastSequence, long lastTimestamp) {
+      this.producerId = producerId;
+      this.producerEpoch = producerEpoch;
+      this.lastSequence = lastSequence;
+      this.lastTimestamp = lastTimestamp;
+    }
+  }
+
+  static class Partition implements JsonObject {
+    final String topic;
+    final int partition;
+    final List<ProducerState> states;
+
+    Partition(String topic, int partition, List<ProducerState> states) {
+      this.topic = topic;
+      this.partition = partition;
+      this.states = states;
+    }
+  }
+
+  static class Partitions implements JsonObject {
+    final List<Partition> partitions;
+
+    Partitions(List<Partition> partitions) {
+      this.partitions = partitions;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/WebService.java b/app/src/main/java/org/astraea/web/WebService.java
index 80fe8b53..a3dd08d6 100644
--- a/app/src/main/java/org/astraea/web/WebService.java
+++ b/app/src/main/java/org/astraea/web/WebService.java
@@ -18,6 +18,7 @@ public class WebService {
     server.createContext("/topics", new TopicHandler(Admin.of(arg.configs())));
     server.createContext("/groups", new GroupHandler(Admin.of(arg.configs())));
     server.createContext("/brokers", new BrokerHandler(Admin.of(arg.configs())));
+    server.createContext("/producers", new ProducerHandler(Admin.of(arg.configs())));
     server.start();
   }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/admin/TopicPartitionTest.java b/app/src/test/java/org/astraea/admin/TopicPartitionTest.java
index 4fea265b..7f8a1151 100644
--- a/app/src/test/java/org/astraea/admin/TopicPartitionTest.java
+++ b/app/src/test/java/org/astraea/admin/TopicPartitionTest.java
@@ -22,4 +22,15 @@ public class TopicPartitionTest {
     Assertions.assertEquals(0, tp0.compareTo(tp0));
     Assertions.assertEquals(tp0, tp0);
   }
+
+  @Test
+  void testString() {
+    Assertions.assertEquals(0, TopicPartition.of("a-a", "0").partition());
+    Assertions.assertThrows(IllegalArgumentException.class, () -> TopicPartition.of("a", "a"));
+
+    Assertions.assertEquals(0, TopicPartition.of("a-a-0").partition());
+    Assertions.assertThrows(IllegalArgumentException.class, () -> TopicPartition.of("a-a"));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> TopicPartition.of("a-a-"));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> TopicPartition.of("aaa"));
+  }
 }
diff --git a/app/src/test/java/org/astraea/web/ProducerHandlerTest.java b/app/src/test/java/org/astraea/web/ProducerHandlerTest.java
new file mode 100644
index 00000000..1f27e98a
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/ProducerHandlerTest.java
@@ -0,0 +1,133 @@
+package org.astraea.web;
+
+import java.util.Map;
+import java.util.Optional;
+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.admin.Admin;
+import org.astraea.producer.Producer;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ProducerHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testListProducers() throws ExecutionException, InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.of(bootstrapServers())) {
+      var handler = new ProducerHandler(admin);
+      producer.sender().topic(topicName).value(new byte[1]).run().toCompletableFuture().get();
+
+      var result =
+          Assertions.assertInstanceOf(
+              ProducerHandler.Partitions.class, handler.get(Optional.empty(), Map.of()));
+      Assertions.assertNotEquals(0, result.partitions.size());
+
+      var partitions =
+          result.partitions.stream()
+              .filter(t -> t.topic.equals(topicName))
+              .collect(Collectors.toUnmodifiableList());
+      Assertions.assertEquals(1, partitions.size());
+      Assertions.assertEquals(topicName, partitions.iterator().next().topic);
+      Assertions.assertEquals(0, partitions.iterator().next().partition);
+    }
+  }
+
+  @Test
+  void testQuerySinglePartition() throws ExecutionException, InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.of(bootstrapServers())) {
+      admin.creator().topic(topicName).numberOfPartitions(2).create();
+      TimeUnit.SECONDS.sleep(2);
+      var handler = new ProducerHandler(admin);
+      producer
+          .sender()
+          .topic(topicName)
+          .partition(0)
+          .value(new byte[1])
+          .run()
+          .toCompletableFuture()
+          .get();
+      producer
+          .sender()
+          .topic(topicName)
+          .partition(1)
+          .value(new byte[1])
+          .run()
+          .toCompletableFuture()
+          .get();
+      TimeUnit.SECONDS.sleep(2);
+
+      Assertions.assertEquals(
+          1,
+          handler
+              .partitions(
+                  Map.of(ProducerHandler.TOPIC_KEY, topicName, ProducerHandler.PARTITION_KEY, "0"))
+              .size());
+
+      var result0 =
+          Assertions.assertInstanceOf(
+              ProducerHandler.Partitions.class,
+              handler.get(
+                  Optional.empty(),
+                  Map.of(
+                      ProducerHandler.TOPIC_KEY, topicName, ProducerHandler.PARTITION_KEY, "0")));
+      Assertions.assertEquals(1, result0.partitions.size());
+      Assertions.assertEquals(topicName, result0.partitions.iterator().next().topic);
+      Assertions.assertEquals(0, result0.partitions.iterator().next().partition);
+
+      Assertions.assertEquals(
+          2, handler.partitions(Map.of(ProducerHandler.TOPIC_KEY, topicName)).size());
+
+      var result1 =
+          Assertions.assertInstanceOf(
+              ProducerHandler.Partitions.class,
+              handler.get(Optional.empty(), Map.of(ProducerHandler.TOPIC_KEY, topicName)));
+      Assertions.assertEquals(2, result1.partitions.size());
+      Assertions.assertEquals(topicName, result1.partitions.iterator().next().topic);
+      Assertions.assertEquals(
+          Set.of(0, 1),
+          result1.partitions.stream().map(p -> p.partition).collect(Collectors.toSet()));
+    }
+  }
+
+  @Test
+  void testPartitions() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers())) {
+      var handler = new ProducerHandler(admin);
+      Assertions.assertEquals(admin.partitions(), handler.partitions(Map.of()));
+      var target = admin.partitions().iterator().next();
+      Assertions.assertEquals(
+          Set.of(target),
+          handler.partitions(
+              Map.of(
+                  ProducerHandler.TOPIC_KEY,
+                  target.topic(),
+                  ProducerHandler.PARTITION_KEY,
+                  String.valueOf(target.partition()))));
+
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              handler.partitions(
+                  Map.of(
+                      ProducerHandler.TOPIC_KEY,
+                      target.topic(),
+                      ProducerHandler.PARTITION_KEY,
+                      "a")));
+
+      admin.creator().topic(topicName).numberOfPartitions(3).create();
+      TimeUnit.SECONDS.sleep(2);
+
+      Assertions.assertEquals(
+          3, handler.partitions(Map.of(ProducerHandler.TOPIC_KEY, topicName)).size());
+    }
+  }
+}