rebase

confluentinc · Sep 25, 2020 · db74090 · db74090
1 parent b635f18
commit db74090
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 9 deletions.
diff --git a/ksqldb-rest-app/src/main/java/io/confluent/ksql/api/impl/QueryEndpoint.java b/ksqldb-rest-app/src/main/java/io/confluent/ksql/api/impl/QueryEndpoint.java
@@ -109,7 +109,7 @@ private QueryPublisher createPullQueryPublisher(
       final long startTimeNanos
   ) {
     final PullQueryResult result = pullQueryExecutor.execute(
-        statement, serviceContext, Optional.of(false), pullQueryMetrics);
+        statement, ImmutableMap.of(), serviceContext, Optional.of(false), pullQueryMetrics);
     pullQueryMetrics.ifPresent(p -> p.recordLatency(startTimeNanos));
     final TableRows tableRows = result.getTableRows();
 

diff --git a/...p/src/main/java/io/confluent/ksql/rest/server/resources/streaming/PullQueryPublisher.java b/...p/src/main/java/io/confluent/ksql/rest/server/resources/streaming/PullQueryPublisher.java
@@ -19,6 +19,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.execution.streams.materialization.Locator.KsqlNode;
 import io.confluent.ksql.parser.tree.Query;
@@ -66,7 +67,7 @@ public synchronized void subscribe(final Subscriber<Collection<StreamedRow>> sub
         subscriber,
         () -> {
           final PullQueryResult result = pullQueryExecutor.execute(
-              query, serviceContext, Optional.of(false), pullQueryMetrics);
+              query, ImmutableMap.of(), serviceContext, Optional.of(false), pullQueryMetrics);
           //Record latency at microsecond scale
           pullQueryMetrics.ifPresent(pullQueryExecutorMetrics -> pullQueryExecutorMetrics
               .recordLatency(startTimeNanos));

diff --git a/...rc/main/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResource.java b/...rc/main/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResource.java
@@ -263,8 +263,8 @@ private EndpointResponse handlePullQuery(
     final ConfiguredStatement<Query> configured = ConfiguredStatement
         .of(statement, SessionConfig.of(ksqlConfig, configOverrides));
 
-    final PullQueryResult result = pullQueryExecutor
-        .execute(configured, serviceContext, isInternalRequest, pullQueryMetrics);
+    final PullQueryResult result = pullQueryExecutor.execute(
+        configured, requestProperties, serviceContext, isInternalRequest, pullQueryMetrics);
     final TableRows tableRows = result.getTableRows();
     final Optional<KsqlHostInfoEntity> host = result.getSourceNode()
         .map(KsqlNode::location)

diff --git a/...rest-app/src/test/java/io/confluent/ksql/rest/server/execution/PullQueryExecutorTest.java b/...rest-app/src/test/java/io/confluent/ksql/rest/server/execution/PullQueryExecutorTest.java
@@ -81,7 +81,7 @@ public void shouldThrowExceptionIfConfigDisabled() {
       final Exception e = assertThrows(
           KsqlStatementException.class,
           () -> pullQueryExecutor.execute(
-              query, engine.getServiceContext(), Optional.empty(),  Optional.empty())
+              query, ImmutableMap.of(), engine.getServiceContext(), Optional.empty(),  Optional.empty())
       );
 
       // Then:

diff --git a/...c/test/java/io/confluent/ksql/rest/server/resources/streaming/PullQueryPublisherTest.java b/...c/test/java/io/confluent/ksql/rest/server/resources/streaming/PullQueryPublisherTest.java
@@ -86,7 +86,7 @@ public void setUp() {
         TIME_NANOS);
 
     PullQueryResult result = new PullQueryResult(entity, Optional.empty());
-    when(pullQueryExecutor.execute(any(), any(), any(), any())).thenReturn(result);
+    when(pullQueryExecutor.execute(any(), any(), any(), any(), any())).thenReturn(result);
     when(entity.getSchema()).thenReturn(SCHEMA);
 
     doAnswer(callRequestAgain()).when(subscriber).onNext(any());
@@ -110,7 +110,7 @@ public void shouldRunQueryWithCorrectParams() {
     subscription.request(1);
 
     // Then:
-    verify(pullQueryExecutor).execute(statement, serviceContext, Optional.of(false), Optional.empty());
+    verify(pullQueryExecutor).execute(statement, ImmutableMap.of(), serviceContext, Optional.of(false), Optional.empty());
   }
 
   @Test
@@ -123,7 +123,7 @@ public void shouldOnlyExecuteOnce() {
 
     // Then:
     verify(subscriber).onNext(any());
-    verify(pullQueryExecutor).execute(statement, serviceContext, Optional.of(false), Optional.empty());
+    verify(pullQueryExecutor).execute(statement, ImmutableMap.of(), serviceContext, Optional.of(false), Optional.empty());
   }
 
   @Test
@@ -158,7 +158,7 @@ public void shouldCallOnErrorOnFailure() {
     // Given:
     givenSubscribed();
     final Throwable e = new RuntimeException("Boom!");
-    when(pullQueryExecutor.execute(any(), any(), any(), any())).thenThrow(e);
+    when(pullQueryExecutor.execute(any(), any(), any(), any(), any())).thenThrow(e);
 
     // When:
     subscription.request(1);