feat(client): Java client with push + pull query support

ksqldb-api-client/pom.xml

@@ -0,0 +1,97 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+
+    <parent>
+        <groupId>io.confluent.ksql</groupId>
+        <artifactId>ksqldb-parent</artifactId>
+        <version>6.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>ksqldb-api-client</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.confluent.ksql</groupId>
+            <artifactId>ksqldb-rest-app</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jdk8</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-core</artifactId>
+            <version>${vertx.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams</artifactId>
+            <version>${reactive-streams.version}</version>
+        </dependency>
+
+        <!-- Required for running tests -->
+
+        <dependency>
+            <groupId>io.confluent.ksql</groupId>
+            <artifactId>ksqldb-rest-app</artifactId>
+            <version>${project.parent.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.confluent.ksql</groupId>
+            <artifactId>ksqldb-test-util</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- Only required to provide LoginModule implementation in tests -->
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-jaas</artifactId>
+            <version>${jetty.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

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

@@ -0,0 +1,76 @@
+/*
+ * 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 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;
+import org.reactivestreams.Publisher;
+
+public interface Client {
+
+  /**
+   * Execute a query (push or pull) and receive the results one row at a time.
+   *
+   * @param sql statement of query to execute.
+   * @return query result.
+   */
+  CompletableFuture<QueryResult> streamQuery(String sql);
+
+  /**
+   * Execute a query (push or pull) and receive the results one row at a time.
+   *
+   * @param sql statement of query to execute.
+   * @param properties query properties.
+   * @return query result.
+   */
+  CompletableFuture<QueryResult> streamQuery(String sql, Map<String, Object> properties);
+
+  /**
+   * Execute a query (push or pull) and receive all result rows together, once the query has
+   * completed.
+   *
+   * @param sql statement of query to execute.
+   * @return query result.
+   */
+  CompletableFuture<List<Row>> executeQuery(String sql);
+
+  /**
+   * Execute a query (push or pull) and receive all result rows together, once the query has
+   * completed.
+   *
+   * @param sql statement of query to execute.
+   * @param properties query properties.
+   * @return query result.
+   */
+  CompletableFuture<List<Row>> executeQuery(String sql, Map<String, Object> properties);
+
+  CompletableFuture<Void> insertInto(String streamName, Map<String, Object> row);
+
+  Publisher<InsertAck> streamInserts(String streamName, Publisher<List<Object>> insertsPublisher);
+
+  void close();
+
+  static Client create(ClientOptions clientOptions) {
+    return new ClientImpl(clientOptions);
+  }
+
+  static Client create(ClientOptions clientOptions, Vertx vertx) {
+    return new ClientImpl(clientOptions, vertx);
+  }
+}

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

@@ -0,0 +1,79 @@
+/*
+ * 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 io.confluent.ksql.api.client.impl.ClientOptionsImpl;
+
+public interface ClientOptions {
+
+  ClientOptions setHost(String host);
+
+  ClientOptions setPort(int port);
+
+  ClientOptions setUseTls(boolean useTls);
+
+  ClientOptions setUseClientAuth(boolean useClientAuth);
+
+  ClientOptions setVerifyHost(boolean verifyHost);
+
+  ClientOptions setTrustAll(boolean trustAll);
+
+  ClientOptions setTrustStore(String trustStorePath);
+
+  ClientOptions setTrustStorePassword(String trustStorePassword);
+
+  ClientOptions setKeyStore(String keyStorePath);
+
+  ClientOptions setKeyStorePassword(String keyStorePassword);
+
+  ClientOptions setBasicAuthCredentials(String username, String password);
+
+  ClientOptions setExecuteQueryMaxResultRows(int maxRows);
+
+  String getHost();
+
+  int getPort();
+
+  boolean isUseTls();
+
+  boolean isUseClientAuth();
+
+  boolean isVerifyHost();
+
+  boolean isTrustAll();
+
+  boolean isUseBasicAuth();
+
+  String getTrustStore();
+
+  String getTrustStorePassword();
+
+  String getKeyStore();
+
+  String getKeyStorePassword();
+
+  String getBasicAuthUsername();
+
+  String getBasicAuthPassword();
+
+  int getExecuteQueryMaxResultRows();
+
+  ClientOptions copy();
+
+  static ClientOptions create() {
+    return new ClientOptionsImpl();
+  }
+}

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

@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+public interface InsertAck {
+
+  int getNum();
+
+}

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

@@ -0,0 +1,55 @@
+/*
+ * 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;
+import java.util.concurrent.TimeUnit;
+import org.reactivestreams.Publisher;
+
+/**
+ * The result of a query (push or pull), streamed one row at time. Records may be consumed by either
+ * subscribing to the publisher or polling (blocking) for one record at a time. These two methods of
+ * consumption are mutually exclusive; only one method may be used (per QueryResult).
+ */
+public interface QueryResult extends Publisher<Row> {
+
+  List<String> columnNames();
+
+  List<String> columnTypes();
+
+  String queryID();
+
+  /**
+   * Block until a row becomes available.
+   *
+   * @return the row.
+   */
+  Row poll();
+
+  /**
+   * Block until a row becomes available or the timeout has elapsed.
+   *
+   * @param timeout amount of to wait for a row. Non-positive values are interpreted as no timeout.
+   * @param timeUnit unit for timeout param.
+   * @return the row, if available; else, null.
+   */
+  Row poll(long timeout, TimeUnit timeUnit);
+
+  boolean isComplete();
+
+  void close();
+
+}