rebase

confluentinc · Feb 27, 2020 · 9cb9792 · 9cb9792
1 parent 736619f
commit 9cb9792
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 362 deletions.
diff --git a/ksql-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlTarget.java b/ksql-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlTarget.java
@@ -30,7 +30,6 @@
 import io.confluent.ksql.rest.entity.KsqlRequest;
 import io.confluent.ksql.rest.entity.LagReportingMessage;
 import io.confluent.ksql.rest.entity.ServerInfo;
-<<<<<<< HEAD
 import io.confluent.ksql.rest.entity.StreamedRow;
 import io.confluent.ksql.util.VertxCompletableFuture;
 import io.vertx.core.Vertx;
@@ -41,12 +40,8 @@
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.net.SocketAddress;
 import java.util.ArrayList;
-import java.util.List;
-=======
-import java.io.InputStream;
-import java.net.SocketTimeoutException;
 import java.util.Collections;
->>>>>>> 0693542d5... added request properties to ksql request
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -225,7 +220,8 @@ private <T> RestResponse<StreamPublisher<T>> executeQueryRequestWithStreamRespon
       final Optional<Long> previousCommandSeqNum,
       final Function<Buffer, T> mapper
   ) {
-    final KsqlRequest ksqlRequest = createKsqlRequest(ksql, Collections.emptyMap(), previousCommandSeqNum);
+    final KsqlRequest ksqlRequest = createKsqlRequest(
+        ksql, Collections.emptyMap(), previousCommandSeqNum);
     final AtomicReference<StreamPublisher<T>> pubRef = new AtomicReference<>();
     return executeSync(HttpMethod.POST, QUERY_PATH, ksqlRequest, resp -> pubRef.get(),
         (resp, vcf) -> {

diff --git a/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientTest.java b/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientTest.java
@@ -54,6 +54,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Base64;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -112,7 +113,7 @@ public void shouldSendKsqlRequest() {
     assertThat(server.getPath(), is("/ksql"));
     assertThat(server.getHeaders().get("Accept"), is("application/json"));
     assertThat(getKsqlRequest(),
-        is(new KsqlRequest(ksql, properties, 123L)));
+        is(new KsqlRequest(ksql, properties, Collections.emptyMap(), 123L)));
   }
 
   @Test
@@ -161,7 +162,7 @@ public void shouldOverrideProperties() {
     target.postKsqlRequest("some ksql", Optional.of(123L));
 
     // Then:
-    assertThat(getKsqlRequest().getStreamsProperties(), is(props));
+    assertThat(getKsqlRequest().getConfigOverrides(), is(props));
   }
 
   @Test
@@ -300,14 +301,15 @@ public void shouldPostQueryRequest() {
 
     // When:
     KsqlTarget target = ksqlClient.target(serverUri);
-    RestResponse<List<StreamedRow>> response = target.postQueryRequest(sql, Optional.of(321L));
+    RestResponse<List<StreamedRow>> response = target.postQueryRequest(
+        sql, Collections.emptyMap(), Optional.of(321L));
 
     // Then:
     assertThat(server.getHttpMethod(), is(HttpMethod.POST));
 
     assertThat(server.getPath(), is("/query"));
     assertThat(server.getHeaders().get("Accept"), is("application/json"));
-    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, 321L)));
+    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, Collections.emptyMap(), 321L)));
     assertThat(response.get(), is(expectedResponse));
   }
 
@@ -330,7 +332,7 @@ public void shouldPostQueryRequestStreamed() throws Exception {
 
     assertThat(server.getPath(), is("/query"));
     assertThat(server.getHeaders().get("Accept"), is("application/json"));
-    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, 321L)));
+    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, Collections.emptyMap(), 321L)));
 
     List<StreamedRow> rows = getElementsFromPublisher(numRows, response.getResponse());
     assertThat(rows, is(expectedResponse));
@@ -355,7 +357,7 @@ public void shouldPostQueryRequestStreamedWithLimit() throws Exception {
 
     assertThat(server.getPath(), is("/query"));
     assertThat(server.getHeaders().get("Accept"), is("application/json"));
-    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, 321L)));
+    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, Collections.emptyMap(), 321L)));
 
     List<StreamedRow> rows = getElementsFromPublisher(numRows + 1, response.getResponse());
     assertThat(rows, is(expectedResponse));
@@ -374,7 +376,7 @@ public void shouldCloseConnectionWhenQueryStreamIsClosed() throws Exception {
         .postQueryRequestStreamed(sql, Optional.of(321L));
 
     // Then:
-    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, 321L)));
+    assertThat(getKsqlRequest(), is(new KsqlRequest(sql, properties, Collections.emptyMap(), 321L)));
 
     // When:
     response.getResponse().close();
@@ -401,7 +403,7 @@ public void shouldExecutePrintTopic() throws Exception {
 
     assertThat(server.getPath(), is("/query"));
     assertThat(server.getHeaders().get("Accept"), is("application/json"));
-    assertThat(getKsqlRequest(), is(new KsqlRequest(command, properties, 123L)));
+    assertThat(getKsqlRequest(), is(new KsqlRequest(command, properties, Collections.emptyMap(), 123L)));
 
     List<String> lines = getElementsFromPublisher(numRows, response.getResponse());
     assertThat(lines, is(expectedResponse));
@@ -420,7 +422,7 @@ public void shouldCloseConnectionWhenPrintTopicPublisherIsClosed() throws Except
         .postPrintTopicRequest(command, Optional.of(123L));
 
     // Then:
-    assertThat(getKsqlRequest(), is(new KsqlRequest(command, properties, 123L)));
+    assertThat(getKsqlRequest(), is(new KsqlRequest(command, properties, Collections.emptyMap(), 123L)));
 
     // When:
     response.getResponse().close();

diff --git a/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientUtilTest.java b/ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientUtilTest.java
@@ -31,6 +31,7 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpClientResponse;
 import io.vertx.core.json.JsonObject;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.function.Function;
@@ -170,15 +171,15 @@ public void shouldSerialiseDeserialise() {
     // Given:
     Map<String, Object> props = new HashMap<>();
     props.put("auto.offset.reset", "latest");
-    KsqlRequest request = new KsqlRequest("some ksql", props, 21345L);
+    KsqlRequest request = new KsqlRequest("some ksql", props, Collections.emptyMap(), 21345L);
 
     // When:
     Buffer buff = KsqlClientUtil.serialize(request);
 
     // Then:
     assertThat(buff, is(notNullValue()));
-    String expectedJson = "{\"ksql\":\"some ksql\",\"streamsProperties\":{\"auto.offset.reset\":\""
-        + "latest\"},\"commandSequenceNumber\":21345}";
+    String expectedJson = "{\"ksql\":\"some ksql\",\"configOverrides\":{\"auto.offset.reset\":\""
+        + "latest\"},\"requestProperties\":{},\"commandSequenceNumber\":21345}";
     assertThat(new JsonObject(buff), is(new JsonObject(expectedJson)));
 
     // When: