Repository: skiptests/astraea (java)

Issue:
Use /status api in waitConnectorInfo method
ref https://github.com/skiptests/astraea/pull/1012#discussion_r1037250993

目前在  ConnectorClient.waitConnectorInfo 中使用`/connectors/(string:name)` 的API 來做 wait 的驗證
期望將他替換為 `/connectors/(string:name)/status`  因為它提供更多的資訊 


Pull request description:
[CONNECTOR] enable to fetch connector status
fix #1199

Patch:
diff --git a/common/src/main/java/org/astraea/common/connector/Builder.java b/common/src/main/java/org/astraea/common/connector/Builder.java
index f64d8704..971e81ed 100644
--- a/common/src/main/java/org/astraea/common/connector/Builder.java
+++ b/common/src/main/java/org/astraea/common/connector/Builder.java
@@ -89,12 +89,19 @@ public class Builder {
       }
 
       @Override
-      public CompletionStage<ConnectorInfo> connector(String name) {
+      public CompletionStage<ConnectorInfo> connectorInfo(String name) {
         return httpExecutor
             .get(getURL(KEY_CONNECTORS + "/" + name), TypeRef.of(ConnectorInfo.class))
             .thenApply(Response::body);
       }
 
+      @Override
+      public CompletionStage<ConnectorStatus> connectorStatus(String name) {
+        return httpExecutor
+            .get(getURL(KEY_CONNECTORS + "/" + name + "/status"), TypeRef.of(ConnectorStatus.class))
+            .thenApply(Response::body);
+      }
+
       @Override
       public CompletionStage<ConnectorInfo> createConnector(
           String name, Map<String, String> config) {
diff --git a/common/src/main/java/org/astraea/common/connector/ConnectorClient.java b/common/src/main/java/org/astraea/common/connector/ConnectorClient.java
index ebffdfa4..14b5e69e 100644
--- a/common/src/main/java/org/astraea/common/connector/ConnectorClient.java
+++ b/common/src/main/java/org/astraea/common/connector/ConnectorClient.java
@@ -45,7 +45,9 @@ public interface ConnectorClient {
 
   CompletionStage<Set<String>> connectorNames();
 
-  CompletionStage<ConnectorInfo> connector(String name);
+  CompletionStage<ConnectorInfo> connectorInfo(String name);
+
+  CompletionStage<ConnectorStatus> connectorStatus(String name);
 
   CompletionStage<ConnectorInfo> createConnector(String name, Map<String, String> config);
 
@@ -56,11 +58,10 @@ public interface ConnectorClient {
   CompletionStage<Set<PluginInfo>> plugins();
 
   default CompletionStage<Boolean> waitConnectorInfo(
-      String connectName, Predicate<ConnectorInfo> predicate, Duration timeout) {
+      String connectName, Predicate<ConnectorStatus> predicate, Duration timeout) {
     return Utils.loop(
         () ->
-            // TODO: 2022-12-01 astraea-1199 Replace by /status api
-            connector(connectName)
+            connectorStatus(connectName)
                 .thenApply(predicate::test)
                 .exceptionally(
                     e -> {
diff --git a/common/src/main/java/org/astraea/common/connector/ConnectorStatus.java b/common/src/main/java/org/astraea/common/connector/ConnectorStatus.java
new file mode 100644
index 00000000..20612f0e
--- /dev/null
+++ b/common/src/main/java/org/astraea/common/connector/ConnectorStatus.java
@@ -0,0 +1,40 @@
+/*
+ * 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.common.connector;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+public class ConnectorStatus {
+
+  private String name;
+  private Map<String, String> connector;
+  private List<TaskStatus> tasks;
+
+  public String name() {
+    return name;
+  }
+
+  public String state() {
+    return Objects.requireNonNull(connector.get("state"));
+  }
+
+  public List<TaskStatus> tasks() {
+    return tasks;
+  }
+}
diff --git a/common/src/main/java/org/astraea/common/connector/TaskStatus.java b/common/src/main/java/org/astraea/common/connector/TaskStatus.java
new file mode 100644
index 00000000..fe14d526
--- /dev/null
+++ b/common/src/main/java/org/astraea/common/connector/TaskStatus.java
@@ -0,0 +1,47 @@
+/*
+ * 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.common.connector;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Optional;
+
+public class TaskStatus {
+
+  private int id;
+  private String state;
+
+  @JsonProperty("worker_id")
+  private String workerId;
+
+  private Optional<String> trace = Optional.empty();
+
+  public int id() {
+    return id;
+  }
+
+  public String state() {
+    return state;
+  }
+
+  public String workerId() {
+    return workerId;
+  }
+
+  public Optional<String> trace() {
+    return trace;
+  }
+}
diff --git a/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java b/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java
index e9caf19c..5942899d 100644
--- a/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java
+++ b/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java
@@ -65,7 +65,7 @@ public class ConnectorNode {
                                         name ->
                                             context
                                                 .connectorClient()
-                                                .connector(name)
+                                                .connectorInfo(name)
                                                 .toCompletableFuture())
                                     .collect(Collectors.toList())))
                     .thenApply(


Test patch:
diff --git a/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java b/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java
index 6d28eccb..020f33de 100644
--- a/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java
+++ b/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java
@@ -33,7 +33,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletionException;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.common.Utils;
@@ -46,53 +46,53 @@ import org.junit.jupiter.api.Test;
 class ConnectorClientTest extends RequireWorkerCluster {
 
   @Test
-  void testInfo() throws ExecutionException, InterruptedException {
+  void testInfo() {
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
-    var info = connectorClient.info().toCompletableFuture().get();
+    var info = connectorClient.info().toCompletableFuture().join();
     assertFalse(isBlank(info.commit()));
     assertFalse(isBlank(info.version()));
     assertFalse(isBlank(info.kafkaClusterId()));
   }
 
   @Test
-  void testConnectors() throws ExecutionException, InterruptedException {
+  void testConnectors() {
     var connectorName = Utils.randomString(10);
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
-    var connectors = connectorClient.connectorNames().toCompletableFuture().get();
+    var connectors = connectorClient.connectorNames().toCompletableFuture().join();
     assertFalse(connectors.stream().anyMatch(x -> x.equals(connectorName)));
 
     connectorClient
         .createConnector(connectorName, getExampleConnector())
         .toCompletableFuture()
-        .get();
-    connectors = connectorClient.connectorNames().toCompletableFuture().get();
+        .join();
+    connectors = connectorClient.connectorNames().toCompletableFuture().join();
     assertTrue(connectors.stream().anyMatch(x -> x.equals(connectorName)));
   }
 
   @Test
-  void testConnectorByName() throws ExecutionException, InterruptedException {
+  void testConnectorByName() {
     var connectorName = Utils.randomString(10);
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
 
-    var executionException =
+    var CompletionException =
         assertThrows(
-            ExecutionException.class,
-            () -> connectorClient.connector(connectorName).toCompletableFuture().get());
-    var exception = getResponseException(executionException);
+            CompletionException.class,
+            () -> connectorClient.connectorInfo(connectorName).toCompletableFuture().join());
+    var exception = getResponseException(CompletionException);
     assertEquals(404, exception.code());
     assertTrue(exception.getMessage().contains(connectorName));
 
     connectorClient
         .createConnector(connectorName, getExampleConnector())
         .toCompletableFuture()
-        .get();
-    var connector = connectorClient.connector(connectorName).toCompletableFuture().get();
+        .join();
+    var connector = connectorClient.connectorInfo(connectorName).toCompletableFuture().join();
     assertEquals(connectorName, connector.name());
     assertExampleConnector(connector.config());
   }
 
   @Test
-  void testCreateConnector() throws ExecutionException, InterruptedException {
+  void testCreateConnector() {
     var connectorName = Utils.randomString(10);
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
     var exampleConnector = new HashMap<>(getExampleConnector());
@@ -102,7 +102,7 @@ class ConnectorClientTest extends RequireWorkerCluster {
         connectorClient
             .createConnector(connectorName, exampleConnector)
             .toCompletableFuture()
-            .get();
+            .join();
     assertEquals(connectorName, createdConnectorInfo.name());
     var config = createdConnectorInfo.config();
     assertEquals("3", config.get("tasks.max"));
@@ -113,8 +113,8 @@ class ConnectorClientTest extends RequireWorkerCluster {
         connectorClient
             .waitConnectorInfo(connectorName, x -> x.tasks().size() > 0, Duration.ofSeconds(10))
             .toCompletableFuture()
-            .get());
-    var connectorInfo = connectorClient.connector(connectorName).toCompletableFuture().get();
+            .join());
+    var connectorInfo = connectorClient.connectorInfo(connectorName).toCompletableFuture().join();
     assertEquals(3, connectorInfo.tasks().size());
     assertTrue(
         connectorInfo.tasks().stream().allMatch(x -> connectorName.equals(x.connectorName())));
@@ -122,7 +122,7 @@ class ConnectorClientTest extends RequireWorkerCluster {
   }
 
   @Test
-  void testUpdateConnector() throws ExecutionException, InterruptedException {
+  void testUpdateConnector() {
     var connectorName = Utils.randomString(10);
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
     var exampleConnector = getExampleConnector();
@@ -131,10 +131,10 @@ class ConnectorClientTest extends RequireWorkerCluster {
         connectorClient
             .createConnector(connectorName, exampleConnector)
             .toCompletableFuture()
-            .get();
+            .join();
     assertEquals("1", connector.config().get("tasks.max"));
     assertEquals("myTopic", connector.config().get("topics"));
-    var connectors = connectorClient.connectorNames().toCompletableFuture().get();
+    var connectors = connectorClient.connectorNames().toCompletableFuture().join();
     assertTrue(connectors.stream().anyMatch(x -> x.equals(connectorName)));
 
     var updateConfig = new HashMap<>(exampleConnector);
@@ -142,60 +142,68 @@ class ConnectorClientTest extends RequireWorkerCluster {
     updateConfig.put("topics", "myTopic2");
 
     connector =
-        connectorClient.updateConnector(connectorName, updateConfig).toCompletableFuture().get();
+        connectorClient.updateConnector(connectorName, updateConfig).toCompletableFuture().join();
     assertEquals("2", connector.config().get("tasks.max"));
     assertEquals("myTopic2", connector.config().get("topics"));
 
     // wait for syncing configs
     Utils.sleep(Duration.ofSeconds(2));
-
-    connector = connectorClient.connector(connectorName).toCompletableFuture().get();
+    connector = connectorClient.connectorInfo(connectorName).toCompletableFuture().join();
     assertEquals("2", connector.config().get("tasks.max"));
     assertEquals("myTopic2", connector.config().get("topics"));
+    assertEquals(2, connector.tasks().size());
+
+    // wait for syncing configs
+    Utils.sleep(Duration.ofSeconds(2));
+    var status = connectorClient.connectorStatus(connectorName).toCompletableFuture().join();
+    assertEquals(connectorName, status.name());
+    assertEquals("RUNNING", status.state());
+    assertEquals(2, status.tasks().size());
+    status.tasks().forEach(t -> assertEquals("RUNNING", t.state()));
   }
 
   @Test
-  void testDeleteConnector() throws ExecutionException, InterruptedException {
+  void testDeleteConnector() {
     var connectorName = Utils.randomString(10);
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
     var exampleConnector = getExampleConnector();
 
-    connectorClient.createConnector(connectorName, exampleConnector).toCompletableFuture().get();
-    var connectors = connectorClient.connectorNames().toCompletableFuture().get();
+    connectorClient.createConnector(connectorName, exampleConnector).toCompletableFuture().join();
+    var connectors = connectorClient.connectorNames().toCompletableFuture().join();
     assertTrue(connectors.stream().anyMatch(x -> x.equals(connectorName)));
 
-    connectorClient.deleteConnector(connectorName).toCompletableFuture().get();
-    connectors = connectorClient.connectorNames().toCompletableFuture().get();
+    connectorClient.deleteConnector(connectorName).toCompletableFuture().join();
+    connectors = connectorClient.connectorNames().toCompletableFuture().join();
     assertFalse(connectors.stream().anyMatch(x -> x.equals(connectorName)));
 
-    var executionException =
+    var CompletionException =
         assertThrows(
-            ExecutionException.class,
-            () -> connectorClient.deleteConnector("unknown").toCompletableFuture().get());
+            CompletionException.class,
+            () -> connectorClient.deleteConnector("unknown").toCompletableFuture().join());
 
-    var exception = getResponseException(executionException);
+    var exception = getResponseException(CompletionException);
 
     // In distribution mode, Request forward to another node will throw 500, otherwise 404.
     assertTrue(exception.getMessage().contains("unknown not found"));
   }
 
   @Test
-  void testPlugin() throws ExecutionException, InterruptedException {
+  void testPlugin() {
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
-    var plugins = connectorClient.plugins().toCompletableFuture().get();
+    var plugins = connectorClient.plugins().toCompletableFuture().join();
     assertTrue(
         plugins.stream()
             .anyMatch(x -> TestTextSourceConnector.class.getName().equals(x.className())));
   }
 
   @Test
-  void testUrls() throws ExecutionException, InterruptedException {
+  void testUrls() {
     var connectorName = Utils.randomString(10);
     var connectorClient = ConnectorClient.builder().urls(new HashSet<>(workerUrls())).build();
     connectorClient
         .createConnector(connectorName, getExampleConnector())
         .toCompletableFuture()
-        .get();
+        .join();
 
     // wait for syncing configs
     Utils.sleep(Duration.ofSeconds(2));
@@ -206,7 +214,7 @@ class ConnectorClientTest extends RequireWorkerCluster {
                 Utils.packException(
                     () -> {
                       var connector =
-                          connectorClient.connector(connectorName).toCompletableFuture().get();
+                          connectorClient.connectorInfo(connectorName).toCompletableFuture().join();
                       assertEquals(connectorName, connector.name());
                     }));
   }
@@ -231,7 +239,7 @@ class ConnectorClientTest extends RequireWorkerCluster {
               .mapToObj(
                   x ->
                       Utils.packException(
-                          () -> connectorClient.connectorNames().toCompletableFuture().get()))
+                          () -> connectorClient.connectorNames().toCompletableFuture().join()))
               .flatMap(Collection::stream)
               .collect(Collectors.toSet());
 
@@ -242,27 +250,30 @@ class ConnectorClientTest extends RequireWorkerCluster {
   }
 
   @Test
-  void testWaitConnectorInfo() throws ExecutionException, InterruptedException {
+  void testWaitConnectorInfo() {
     var connectorName = Utils.randomString(10);
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
     var exampleConnector = new HashMap<>(getExampleConnector());
 
-    connectorClient.createConnector(connectorName, exampleConnector).toCompletableFuture().get();
+    connectorClient.createConnector(connectorName, exampleConnector).toCompletableFuture().join();
     assertTrue(
         connectorClient
             .waitConnectorInfo(connectorName, x -> x.tasks().size() > 0, Duration.ofSeconds(10))
             .toCompletableFuture()
-            .get());
+            .join());
 
     var errConnectorName = Utils.randomString(10);
     exampleConnector.put("connector.class", TestErrorSourceConnector.class.getName());
-    connectorClient.createConnector(errConnectorName, exampleConnector).toCompletableFuture().get();
+    connectorClient
+        .createConnector(errConnectorName, exampleConnector)
+        .toCompletableFuture()
+        .join();
 
     assertFalse(
         connectorClient
             .waitConnectorInfo(errConnectorName, x -> x.tasks().size() > 0, Duration.ofSeconds(10))
             .toCompletableFuture()
-            .get());
+            .join());
   }
 
   private static class Server implements AutoCloseable {
@@ -295,10 +306,10 @@ class ConnectorClientTest extends RequireWorkerCluster {
     }
   }
 
-  private HttpRequestException getResponseException(ExecutionException executionException) {
-    executionException.printStackTrace();
-    assertEquals(HttpRequestException.class, executionException.getCause().getClass());
-    return (HttpRequestException) executionException.getCause();
+  private HttpRequestException getResponseException(Exception CompletionException) {
+    CompletionException.printStackTrace();
+    assertEquals(HttpRequestException.class, CompletionException.getCause().getClass());
+    return (HttpRequestException) CompletionException.getCause();
   }
 
   private void assertExampleConnector(Map<String, String> config) {
diff --git a/connector/src/test/java/org/astraea/connector/ConnectorTest.java b/connector/src/test/java/org/astraea/connector/ConnectorTest.java
index 5d42d79a..b90a1706 100644
--- a/connector/src/test/java/org/astraea/connector/ConnectorTest.java
+++ b/connector/src/test/java/org/astraea/connector/ConnectorTest.java
@@ -70,7 +70,8 @@ public class ConnectorTest extends RequireWorkerCluster {
 
     // wait for sync
     Utils.sleep(Duration.ofSeconds(3));
-    Assertions.assertEquals(3, client.connector(name).toCompletableFuture().join().tasks().size());
+    Assertions.assertEquals(
+        3, client.connectorInfo(name).toCompletableFuture().join().tasks().size());
     Assertions.assertEquals(1, MySource.INIT_COUNT.get());
     Assertions.assertEquals(3, MySourceTask.INIT_COUNT.get());
     Assertions.assertEquals(0, MySource.CLOSE_COUNT.get());