fix: fix test failures

confluentinc · Aug 13, 2021 · 0cce482 · 0cce482
1 parent 29fcf5b
commit 0cce482
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 4 deletions.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/physical/scalablepush/PushPhysicalPlan.java b/ksqldb-engine/src/main/java/io/confluent/ksql/physical/scalablepush/PushPhysicalPlan.java
@@ -99,7 +99,6 @@ private void maybeNext(final Publisher publisher) {
 
   private boolean isErrored(final Publisher publisher) {
     if (dataSourceOperator.droppedRows()) {
-      System.out.println("dropped rows");
       closeInternal();
       publisher.reportDroppedRows();
       return true;

diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/util/PersistentQueryMetadataImpl.java b/ksqldb-engine/src/main/java/io/confluent/ksql/util/PersistentQueryMetadataImpl.java
@@ -116,10 +116,14 @@ public void onError(final QueryMetadata queryMetadata, final QueryError error) {
 
           @Override
           public void onStateChange(final QueryMetadata queryMetadata, final State before,
-              final State after) { }
+              final State after) {
+            listener.onStateChange(queryMetadata, before, after);
+          }
 
           @Override
-          public void onClose(final QueryMetadata queryMetadata) { }
+          public void onClose(final QueryMetadata queryMetadata) {
+            listener.onClose(queryMetadata);
+          }
         }
     );
     this.sinkDataSource = requireNonNull(sinkDataSource, "sinkDataSource");

diff --git a/...db-engine/src/test/java/io/confluent/ksql/physical/scalablepush/PushPhysicalPlanTest.java b/...db-engine/src/test/java/io/confluent/ksql/physical/scalablepush/PushPhysicalPlanTest.java
@@ -97,6 +97,7 @@ public void shouldStopOnDroppedRows() throws InterruptedException {
         scalablePushRegistry, pushDataSourceOperator, context);
     doNothing().when(pushDataSourceOperator).setNewRowCallback(runnableCaptor.capture());
     when(pushDataSourceOperator.droppedRows()).thenReturn(false, true);
+//    when(pushDataSourceOperator.hasError()).thenReturn(false, false);
 
     final BufferedPublisher<List<?>> publisher = pushPhysicalPlan.execute();
     final TestSubscriber<List<?>> subscriber = new TestSubscriber<>();
@@ -146,7 +147,7 @@ public void shouldStopOnHasError() throws InterruptedException {
 
       context.owner().cancelTimer(timerId);
     });
-//
+
     while (subscriber.getError() == null) {
       Thread.sleep(100);
     }