feat(client): support admin operations in Java client (#5671)

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/Client.java

@@ -17,6 +17,7 @@
 
 import io.confluent.ksql.api.client.impl.ClientImpl;
 import io.vertx.core.Vertx;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
@@ -92,6 +93,36 @@ public interface Client {
    */
   CompletableFuture<Void> terminatePushQuery(String queryId);
 
+  /**
+   * Returns the list of ksqlDB streams from the ksqlDB server's metastore.
+   *
+   * <p>If a non-200 response is received from the server, the {@code CompletableFuture} will be
+   * failed.
+   *
+   * @return list of streams
+   */
+  CompletableFuture<List<StreamInfo>> listStreams();
+
+  /**
+   * Returns the list of ksqlDB tables from the ksqlDB server's metastore
+   *
+   * <p>If a non-200 response is received from the server, the {@code CompletableFuture} will be
+   * failed.
+   *
+   * @return list of tables
+   */
+  CompletableFuture<List<TableInfo>> listTables();
+
+  /**
+   * Returns the list of Kafka topics available for use with ksqlDB.
+   *
+   * <p>If a non-200 response is received from the server, the {@code CompletableFuture} will be
+   * failed.
+   *
+   * @return list of topics
+   */
+  CompletableFuture<List<TopicInfo>> listTopics();
+
   /**
    * Closes the underlying HTTP client.
    */

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/StreamInfo.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.api.client;
+
+/**
+ * Metadata for a ksqlDB stream.
+ */
+public interface StreamInfo {
+
+  /**
+   * @return the name of this stream
+   */
+  String getName();
+
+  /**
+   * @return the name of the Kafka topic underlying this ksqlDB stream
+   */
+  String getTopic();
+
+  /**
+   * @return the format of the data in this stream
+   */
+  String getFormat();
+
+}

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/TableInfo.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.api.client;
+
+/**
+ * Metadata for a ksqlDB table.
+ */
+public interface TableInfo {
+
+  /**
+   * @return the name of this table
+   */
+  String getName();
+
+  /**
+   * @return the name of the Kafka topic underlying this ksqlDB table
+   */
+  String getTopic();
+
+  /**
+   * @return the format of the data in this table
+   */
+  String getFormat();
+
+  /**
+   * @return whether this ksqlDB table is windowed
+   */
+  boolean isWindowed();
+
+}

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/TopicInfo.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.api.client;
+
+import java.util.List;
+
+/**
+ * Metadata for a Kafka topic available for use with ksqlDB.
+ */
+public interface TopicInfo {
+
+  /**
+   * @return the name of this topic
+   */
+  String getName();
+
+  /**
+   * @return the number of partitions for this topic
+   */
+  int getPartitions();
+
+  /**
+   * Returns the number of replicas for each topic partition.
+   *
+   * @return a list with size equal to the number of partitions. Each element is the number of
+   *         replicas for the partition corresponding to the list index.
+   */
+  List<Integer> getReplicasPerPartition();
+
+}

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/AdminResponseHandlers.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.api.client.impl;
+
+import io.confluent.ksql.api.client.StreamInfo;
+import io.confluent.ksql.api.client.TableInfo;
+import io.confluent.ksql.api.client.TopicInfo;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.JsonObject;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+final class AdminResponseHandlers {
+
+  private AdminResponseHandlers() {
+  }
+
+  static void handleListStreamsResponse(
+      final JsonObject streamsListEntity,
+      final CompletableFuture<List<StreamInfo>> cf
+  ) {
+    try {
+      final JsonArray streams = streamsListEntity.getJsonArray("streams");
+      cf.complete(streams.stream()
+          .map(o -> (JsonObject) o)
+          .map(o -> new StreamInfoImpl(
+              o.getString("name"),
+              o.getString("topic"),
+              o.getString("format")))
+          .collect(Collectors.toList())
+      );
+    } catch (Exception e) {
+      cf.completeExceptionally(new IllegalStateException(
+          "Unexpected server response format. Response: " + streamsListEntity));
+    }
+  }
+
+  static void handleListTablesResponse(
+      final JsonObject tablesListEntity,
+      final CompletableFuture<List<TableInfo>> cf
+  ) {
+    try {
+      final JsonArray tables = tablesListEntity.getJsonArray("tables");
+      cf.complete(tables.stream()
+          .map(o -> (JsonObject) o)
+          .map(o -> new TableInfoImpl(
+              o.getString("name"),
+              o.getString("topic"),
+              o.getString("format"),
+              o.getBoolean("isWindowed")))
+          .collect(Collectors.toList())
+      );
+    } catch (Exception e) {
+      cf.completeExceptionally(new IllegalStateException(
+          "Unexpected server response format. Response: " + tablesListEntity));
+    }
+  }
+
+  static void handleListTopicsResponse(
+      final JsonObject kafkaTopicsListEntity,
+      final CompletableFuture<List<TopicInfo>> cf
+  ) {
+    try {
+      final JsonArray topics = kafkaTopicsListEntity.getJsonArray("topics");
+      cf.complete(topics.stream()
+          .map(o -> (JsonObject) o)
+          .map(o -> {
+            final List<Integer> replicaInfo = o.getJsonArray("replicaInfo").stream()
+                .map(v -> (Integer)v)
+                .collect(Collectors.toList());
+            return new TopicInfoImpl(
+                o.getString("name"),
+                replicaInfo.size(),
+                replicaInfo);
+          })
+          .collect(Collectors.toList())
+      );
+    } catch (Exception e) {
+      cf.completeExceptionally(new IllegalStateException(
+          "Unexpected server response format. Response: " + kafkaTopicsListEntity));
+    }
+  }
+
+}