fix feedback

confluentinc · Feb 27, 2020 · 6997894 · 6997894
1 parent 2ec6f22
commit 6997894
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 2 deletions.
diff --git a/ksql-engine/src/main/java/io/confluent/ksql/util/TransientQueryMetadata.java b/ksql-engine/src/main/java/io/confluent/ksql/util/TransientQueryMetadata.java
@@ -104,8 +104,10 @@ public void close(final boolean cleanUp) {
     // write to the blocking queue, otherwise super.close call can deadlock:
     rowQueue.close();
 
-    // Now safe to close:
-    super.close(cleanUp);
+    // Now safe to close
+    // We always pass cleanUp true, since transient queries should always have
+    // their internal topics cleaned up.
+    super.close(true);
     isRunning.set(false);
   }
 }
diff --git a/ksql-engine/src/test/java/io/confluent/ksql/util/TransientQueryMetadataTest.java b/ksql-engine/src/test/java/io/confluent/ksql/util/TransientQueryMetadataTest.java
@@ -17,6 +17,7 @@
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.verify;
 
 import io.confluent.ksql.name.SourceName;
 import io.confluent.ksql.query.BlockingRowQueue;
@@ -87,4 +88,22 @@ public void shouldCloseQueueBeforeTopologyToAvoidDeadLock() {
     inOrder.verify(rowQueue).close();
     inOrder.verify(kafkaStreams).close(any());
   }
+
+  @Test
+  public void shouldCloseWithCleanUpTrue() {
+    // When:
+    query.close(true);
+
+    // Then:
+    verify(closeCallback).accept(query, true);
+  }
+
+  @Test
+  public void shouldCloseWithCleanUpTrueEvenIfPassedFalse() {
+    // When:
+    query.close(false);
+
+    // Then:
+    verify(closeCallback).accept(query, true);
+  }
 }