Repository: skiptests/astraea
Pull request #406 (2022-06-12 16:40:04)

Add transaction web APIs
新增web apis用來查詢transaction state


Patch:
diff --git a/app/src/main/java/org/astraea/app/web/GroupHandler.java b/app/src/main/java/org/astraea/app/web/GroupHandler.java
index 670144d8..48905eaa 100644
--- a/app/src/main/java/org/astraea/app/web/GroupHandler.java
+++ b/app/src/main/java/org/astraea/app/web/GroupHandler.java
@@ -87,14 +87,14 @@ public class GroupHandler implements Handler {
   }
 
   static class OffsetProgress implements JsonObject {
-    final String topicName;
+    final String topic;
     final int partitionId;
     final long earliest;
     final long current;
     final long latest;
 
     OffsetProgress(String topicName, int partitionId, long earliest, long current, long latest) {
-      this.topicName = topicName;
+      this.topic = topicName;
       this.partitionId = partitionId;
       this.earliest = earliest;
       this.current = current;
diff --git a/app/src/main/java/org/astraea/app/web/ProducerHandler.java b/app/src/main/java/org/astraea/app/web/ProducerHandler.java
index be3dd789..44868119 100644
--- a/app/src/main/java/org/astraea/app/web/ProducerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/ProducerHandler.java
@@ -16,12 +16,14 @@
  */
 package org.astraea.app.web;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.ProducerState;
 import org.astraea.app.admin.TopicPartition;
 
 class ProducerHandler implements Handler {
@@ -47,23 +49,10 @@ class ProducerHandler implements Handler {
   }
 
   @Override
-  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+  public Partitions 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())))
+            .map(e -> new Partition(e.getKey(), e.getValue()))
             .collect(Collectors.toUnmodifiableList());
     return new Partitions(topics);
   }
@@ -75,11 +64,11 @@ class ProducerHandler implements Handler {
     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;
+    ProducerState(org.astraea.app.admin.ProducerState state) {
+      this.producerId = state.producerId();
+      this.producerEpoch = state.producerEpoch();
+      this.lastSequence = state.lastSequence();
+      this.lastTimestamp = state.lastTimestamp();
     }
   }
 
@@ -88,10 +77,13 @@ class ProducerHandler implements Handler {
     final int partition;
     final List<ProducerState> states;
 
-    Partition(String topic, int partition, List<ProducerState> states) {
-      this.topic = topic;
-      this.partition = partition;
-      this.states = states;
+    Partition(
+        org.astraea.app.admin.TopicPartition tp,
+        Collection<org.astraea.app.admin.ProducerState> states) {
+      this.topic = tp.topic();
+      this.partition = tp.partition();
+      this.states =
+          states.stream().map(ProducerState::new).collect(Collectors.toUnmodifiableList());
     }
   }
 
diff --git a/app/src/main/java/org/astraea/app/web/TransactionHandler.java b/app/src/main/java/org/astraea/app/web/TransactionHandler.java
new file mode 100644
index 00000000..d6bbd5a2
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/web/TransactionHandler.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.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.app.admin.Admin;
+
+class TransactionHandler implements Handler {
+
+  private final Admin admin;
+
+  TransactionHandler(Admin admin) {
+    this.admin = admin;
+  }
+
+  @Override
+  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+    var transactions =
+        admin.transactions(Handler.compare(admin.transactionIds(), target)).entrySet().stream()
+            .map(e -> new Transaction(e.getKey(), e.getValue()))
+            .collect(Collectors.toUnmodifiableList());
+    if (target.isPresent() && transactions.size() == 1) return transactions.get(0);
+    return new Transactions(transactions);
+  }
+
+  static class TopicPartition implements JsonObject {
+    final String topic;
+    final int partition;
+
+    TopicPartition(org.astraea.app.admin.TopicPartition tp) {
+      this.topic = tp.topic();
+      this.partition = tp.partition();
+    }
+  }
+
+  static class Transaction implements JsonObject {
+    final String id;
+    final int coordinatorId;
+    final String state;
+    final long producerId;
+    final int producerEpoch;
+    final long transactionTimeoutMs;
+    final Set<TopicPartition> topicPartitions;
+
+    Transaction(String id, org.astraea.app.admin.Transaction transaction) {
+      this.id = id;
+      this.coordinatorId = transaction.coordinatorId();
+      this.state = transaction.state().toString();
+      this.producerId = transaction.producerId();
+      this.producerEpoch = transaction.producerEpoch();
+      this.transactionTimeoutMs = transaction.transactionTimeoutMs();
+      this.topicPartitions =
+          transaction.topicPartitions().stream()
+              .map(TopicPartition::new)
+              .collect(Collectors.toSet());
+    }
+  }
+
+  static class Transactions implements JsonObject {
+    final List<Transaction> transactions;
+
+    Transactions(List<Transaction> transactions) {
+      this.transactions = transactions;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/web/WebService.java b/app/src/main/java/org/astraea/app/web/WebService.java
index 5ba51f8b..6e32b708 100644
--- a/app/src/main/java/org/astraea/app/web/WebService.java
+++ b/app/src/main/java/org/astraea/app/web/WebService.java
@@ -37,6 +37,7 @@ public class WebService {
     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.createContext("/transaction", new TransactionHandler(Admin.of(arg.configs())));
     server.start();
   }
 
diff --git a/docs/web_server/web_api_groups_chinese.md b/docs/web_server/web_api_groups_chinese.md
index 1c2b569b..7b92c6c6 100644
--- a/docs/web_server/web_api_groups_chinese.md
+++ b/docs/web_server/web_api_groups_chinese.md
@@ -19,7 +19,7 @@ JSON Response 範例
 - `clientId`: consumer 送出請求時使用的名稱，使用者可以自定義，預設值為隨機字串
 - `host`: consumer 運行的位址
 - `offsetProgress`: consumer 目前消化資料的進度
-  - `topicName`: consumer 提取資料的 topic 名稱
+  - `topic`: consumer 提取資料的 topic 名稱
   - `partitionId`: consumer 提取資料的 partition id
   - `earliest`: partition 現存的資料最早的紀錄位置
   - `current`: consumer 於 partition 目前處理的數據位置
@@ -40,7 +40,7 @@ JSON Response 範例
           "host": "/172.17.0.1",
           "offsetProgress": [
             {
-              "topicName": "test1",
+              "topic": "test1",
               "partitionId": 0,
               "earliest": 0,
               "current": 2,
@@ -76,7 +76,7 @@ JSON Response 範例
       "host": "/172.17.0.1",
       "offsetProgress": [
         {
-          "topicName": "test1",
+          "topic": "test1",
           "partitionId": 0,
           "earliest": 0,
           "current": 0,


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index 41a6d319..ea5625ab 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -18,6 +18,7 @@ package org.astraea.app.admin;
 
 import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
@@ -794,18 +795,45 @@ public class AdminTest extends RequireBrokerCluster {
 
   @Test
   void testTransactionIds() throws ExecutionException, InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer =
+            Producer.builder().bootstrapServers(bootstrapServers()).buildTransactional()) {
+      Assertions.assertTrue(producer.transactional());
+      producer.sender().key(new byte[10]).topic(topicName).run().toCompletableFuture().get();
 
+      Assertions.assertTrue(admin.transactionIds().contains(producer.transactionId().get()));
+
+      var transaction = admin.transactions().get(producer.transactionId().get());
+      Assertions.assertNotNull(transaction);
+      Assertions.assertEquals(TransactionState.COMPLETE_COMMIT, transaction.state());
+      Assertions.assertEquals(0, transaction.topicPartitions().size());
+    }
+  }
+
+  @Test
+  void testTransactionIdsWithMultiPuts() throws ExecutionException, InterruptedException {
+    var topicName = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers());
         var producer =
             Producer.builder().bootstrapServers(bootstrapServers()).buildTransactional()) {
       Assertions.assertTrue(producer.transactional());
-      producer.sender().key(new byte[10]).topic("topic").run().toCompletableFuture().get();
+      IntStream.range(0, 10)
+          .forEach(
+              index ->
+                  producer
+                      .sender()
+                      .key(String.valueOf(index).getBytes(StandardCharsets.UTF_8))
+                      .topic(topicName)
+                      .run());
+      producer.flush();
 
       Assertions.assertTrue(admin.transactionIds().contains(producer.transactionId().get()));
 
       var transaction = admin.transactions().get(producer.transactionId().get());
       Assertions.assertNotNull(transaction);
       Assertions.assertEquals(TransactionState.COMPLETE_COMMIT, transaction.state());
+      Assertions.assertEquals(0, transaction.topicPartitions().size());
     }
   }
 }
diff --git a/app/src/test/java/org/astraea/app/web/TransactionHandlerTest.java b/app/src/test/java/org/astraea/app/web/TransactionHandlerTest.java
new file mode 100644
index 00000000..b8aff05f
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/web/TransactionHandlerTest.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.web;
+
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.common.Utils;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TransactionHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testListTransactions() throws ExecutionException, InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer =
+            Producer.builder().bootstrapServers(bootstrapServers()).buildTransactional()) {
+      var handler = new TransactionHandler(admin);
+      producer.sender().topic(topicName).value(new byte[1]).run().toCompletableFuture().get();
+
+      var result =
+          Assertions.assertInstanceOf(
+              TransactionHandler.Transactions.class, handler.get(Optional.empty(), Map.of()));
+      var transaction =
+          result.transactions.stream()
+              .filter(t -> t.id.equals(producer.transactionId().get()))
+              .findFirst()
+              .get();
+      Assertions.assertEquals(0, transaction.topicPartitions.size());
+    }
+  }
+
+  @Test
+  void testQueryTransactionId() throws ExecutionException, InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer =
+            Producer.builder().bootstrapServers(bootstrapServers()).buildTransactional()) {
+      var handler = new TransactionHandler(admin);
+      producer.sender().topic(topicName).value(new byte[1]).run().toCompletableFuture().get();
+
+      var transaction =
+          Assertions.assertInstanceOf(
+              TransactionHandler.Transaction.class,
+              handler.get(Optional.of(producer.transactionId().get()), Map.of()));
+
+      Assertions.assertEquals(0, transaction.topicPartitions.size());
+    }
+  }
+
+  @Test
+  void queryNonexistentTransactionId() throws ExecutionException, InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer =
+            Producer.builder().bootstrapServers(bootstrapServers()).buildTransactional()) {
+      var handler = new TransactionHandler(admin);
+      producer.sender().topic(topicName).value(new byte[1]).run().toCompletableFuture().get();
+
+      Assertions.assertThrows(
+          NoSuchElementException.class,
+          () -> handler.get(Optional.of(Utils.randomString(10)), Map.of()));
+    }
+  }
+}