Repository: skiptests/astraea
Pull request #367 (2022-05-26 12:07:30)

Enable to show consuemr/producer connected to topics
The API `/pipelines` can show the producer/consumer connected to topics.
```json
{
    "topicPartitions": [
        {
            "topic": "testPerformance-1653542175343",
            "partition": 0,
            "from": [
                {
                    "producerId": 6,
                    "producerEpoch": 0,
                    "lastSequence": 129638,
                    "lastTimestamp": 1653542192678
                },
                {
                    "producerId": 7,
                    "producerEpoch": 0,
                    "lastSequence": 129637,
                    "lastTimestamp": 1653542192730
                },
                {
                    "producerId": 8,
                    "producerEpoch": 0,
                    "lastSequence": 129244,
                    "lastTimestamp": 1653542192736
                }
            ],
            "to": [
                {
                    "groupId": "groupId-1653542175709",
                    "memberId": "consumer-groupId-1653542175709-1-c1f578db-5c84-4c06-863e-342c0c3fe654",
                    "clientId": "consumer-groupId-1653542175709-1",
                    "host": "/172.17.0.1"
                }
            ]
        },
        {
            "topic": "testPerformance-1653542175343",
            "partition": 1,
            "from": [
                {
                    "producerId": 6,
                    "producerEpoch": 0,
                    "lastSequence": 130143,
                    "lastTimestamp": 1653542193087
                },
                {
                    "producerId": 7,
                    "producerEpoch": 0,
                    "lastSequence": 129273,
                    "lastTimestamp": 1653542193048
                },
                {
                    "producerId": 8,
                    "producerEpoch": 0,
                    "lastSequence": 129139,
                    "lastTimestamp": 1653542193044
                }
            ],
            "to": [
                {
                    "groupId": "groupId-1653542175709",
                    "memberId": "consumer-groupId-1653542175709-2-36261e98-94da-4fa2-bd26-dee5fb08eeec",
                    "clientId": "consumer-groupId-1653542175709-2",
                    "host": "/172.17.0.1"
                }
            ]
        },
        {
            "topic": "testPerformance-1653542175343",
            "partition": 2,
            "from": [
                {
                    "producerId": 6,
                    "producerEpoch": 0,
                    "lastSequence": 129516,
                    "lastTimestamp": 1653542193295
                },
                {
                    "producerId": 7,
                    "producerEpoch": 0,
                    "lastSequence": 129262,
                    "lastTimestamp": 1653542193313
                },
                {
                    "producerId": 8,
                    "producerEpoch": 0,
                    "lastSequence": 128636,
                    "lastTimestamp": 1653542193280
                }
            ],
            "to": [
                {
                    "groupId": "groupId-1653542175709",
                    "memberId": "consumer-groupId-1653542175709-3-bc5b5686-3391-4ae7-9fce-8f06be20dcc9",
                    "clientId": "consumer-groupId-1653542175709-3",
                    "host": "/172.17.0.1"
                }
            ]
        }
    ]
}
```


Patch:
diff --git a/app/src/main/java/org/astraea/web/PipelineHandler.java b/app/src/main/java/org/astraea/web/PipelineHandler.java
new file mode 100644
index 00000000..851f15ce
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/PipelineHandler.java
@@ -0,0 +1,121 @@
+package org.astraea.web;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.astraea.admin.Admin;
+import org.astraea.admin.Member;
+import org.astraea.admin.ProducerState;
+
+class PipelineHandler implements Handler {
+
+  static final String ACTIVE_KEY = "active";
+
+  private final Admin admin;
+
+  PipelineHandler(Admin admin) {
+    this.admin = admin;
+  }
+
+  @Override
+  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+    var tps =
+        topicPartitions(admin).stream()
+            .filter(filter(queries))
+            .collect(Collectors.toUnmodifiableList());
+    return new TopicPartitions(tps);
+  }
+
+  static Predicate<TopicPartition> filter(Map<String, String> queries) {
+    var flag = queries.get(ACTIVE_KEY);
+    // remove the topic-partitions having no producers/consumers
+    if (flag != null && flag.equalsIgnoreCase("true"))
+      return tp -> !tp.to.isEmpty() || !tp.from.isEmpty();
+    // remove the topic-partitions having producers/consumers
+    if (flag != null && flag.equalsIgnoreCase("false"))
+      return tp -> tp.to.isEmpty() && tp.from.isEmpty();
+    return tp -> true;
+  }
+
+  static Collection<TopicPartition> topicPartitions(Admin admin) {
+    var result =
+        admin.partitions().stream()
+            .collect(
+                Collectors.toMap(
+                    Function.identity(), tp -> new TopicPartition(tp.topic(), tp.partition())));
+    admin
+        .consumerGroups()
+        .values()
+        .forEach(
+            cg ->
+                cg.assignment()
+                    .forEach(
+                        (m, tps) ->
+                            tps.stream()
+                                // it could return new topic partition, so we have to remove them.
+                                .filter(result::containsKey)
+                                .forEach(tp -> result.get(tp).to.add(new Consumer(m)))));
+    admin
+        .producerStates(result.keySet())
+        .forEach((tp, p) -> p.forEach(s -> result.get(tp).from.add(new Producer(s))));
+    return result.values().stream()
+        .sorted(
+            Comparator.comparing((TopicPartition tp) -> tp.topic).thenComparing(tp -> tp.partition))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  static class Producer implements JsonObject {
+    final long producerId;
+    final int producerEpoch;
+    final int lastSequence;
+    final long lastTimestamp;
+
+    Producer(ProducerState state) {
+      this.producerId = state.producerId();
+      this.producerEpoch = state.producerEpoch();
+      this.lastSequence = state.lastSequence();
+      this.lastTimestamp = state.lastTimestamp();
+    }
+  }
+
+  static class Consumer implements JsonObject {
+    final String groupId;
+    final String memberId;
+    final String groupInstanceId;
+    final String clientId;
+    final String host;
+
+    Consumer(Member member) {
+      this.groupId = member.groupId();
+      this.memberId = member.memberId();
+      this.groupInstanceId = member.groupInstanceId().orElse(null);
+      this.clientId = member.clientId();
+      this.host = member.host();
+    }
+  }
+
+  static class TopicPartition implements JsonObject {
+    final String topic;
+    final int partition;
+    final Collection<Producer> from = new ArrayList<>();
+    final Collection<Consumer> to = new ArrayList<>();
+
+    TopicPartition(String topic, int partition) {
+      this.topic = topic;
+      this.partition = partition;
+    }
+  }
+
+  static class TopicPartitions implements JsonObject {
+    final Collection<TopicPartition> topicPartitions;
+
+    TopicPartitions(Collection<TopicPartition> topicPartitions) {
+      this.topicPartitions = topicPartitions;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/WebService.java b/app/src/main/java/org/astraea/web/WebService.java
index d0821feb..c47cf379 100644
--- a/app/src/main/java/org/astraea/web/WebService.java
+++ b/app/src/main/java/org/astraea/web/WebService.java
@@ -20,6 +20,7 @@ public class WebService {
     server.createContext("/brokers", new BrokerHandler(Admin.of(arg.configs())));
     server.createContext("/producers", new ProducerHandler(Admin.of(arg.configs())));
     server.createContext("/quotas", new QuotaHandler(Admin.of(arg.configs())));
+    server.createContext("/pipelines", new PipelineHandler(Admin.of(arg.configs())));
     server.start();
   }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/web/PipelineHandlerTest.java b/app/src/test/java/org/astraea/web/PipelineHandlerTest.java
new file mode 100644
index 00000000..c181d6ed
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/PipelineHandlerTest.java
@@ -0,0 +1,54 @@
+package org.astraea.web;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+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;
+import org.mockito.Mockito;
+
+public class PipelineHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testQueries() {
+    // this topic-partition has no producers/consumers
+    var tp = new PipelineHandler.TopicPartition("g", 1);
+    Assertions.assertTrue(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "false")).test(tp));
+    Assertions.assertFalse(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "true")).test(tp));
+    Assertions.assertTrue(PipelineHandler.filter(Map.of()).test(tp));
+
+    // add some producers/consumers
+    tp.from.add(Mockito.mock(PipelineHandler.Producer.class));
+    Assertions.assertFalse(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "false")).test(tp));
+    Assertions.assertTrue(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "true")).test(tp));
+    Assertions.assertTrue(PipelineHandler.filter(Map.of()).test(tp));
+  }
+
+  @Test
+  void testGetPipeline() throws InterruptedException, ExecutionException {
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(1).create();
+      TimeUnit.SECONDS.sleep(2);
+      try (var producer = Producer.of(bootstrapServers())) {
+        producer.sender().topic(topic).value(new byte[10]).run().toCompletableFuture().get();
+        var handler = new PipelineHandler(admin);
+        var response =
+            Assertions.assertInstanceOf(
+                PipelineHandler.TopicPartitions.class, handler.get(Optional.empty(), Map.of()));
+        Assertions.assertNotEquals(0, response.topicPartitions.size());
+        Assertions.assertEquals(
+            1, response.topicPartitions.stream().filter(t -> t.topic.equals(topic)).count());
+      }
+    }
+  }
+}