KSQL sometimes deletes internal topics when the server shuts down

[rodesai]

When KSQL shuts down, it closes all the queries in the engine. One of the steps of closing a query is to delete the internal topics.

To Reproduce
Start KSQL and run a query, then shut KSQL down (presumably by sending a term signal).

Expected behavior
KSQL exits cleanly but leaves all internal topics in tact

Actual behaviour
KSQL deletes the internal topics (and therefore their data)

[apurvam]

What do you mean deletes internal topics? For persistent queries, or just for transient queries?

[rodesai]

for persistent queries - so it deletes all the repartition topics and changelogs. This is racy, which might explain why we don't hit this in dev environments. Adding some details now.

[apurvam]

thanks. Would be good to understand when it was introduced.

[rodesai]

So the underlying issue is that the shutdown handling logic iterates over all the running queries and terminates them (calls QueryMetadata.close) and the engine makes no distinction between terminating in the context of a shutdown vs in response to a user's statement. And so it tries to delete every query's internal topics (see KsqlEngine.close).

It looks like maybe this bug is being hidden by a race condition, which is why when I tried to reproduce by just shutting the server down by sending a SIGTERM, I didn't see the internal topics deleted.

Let's assume I've started KSQL and run 1 query.

When the JVM gets the SIGTERM, 2 threads react to it:

Jetty's ShutdownThread (which register's with the JVM as a shutdown handler), which calls ApplicationServer.doStop which eventually invokes KsqlRestApplication.triggerShutdown

main, which also calls KsqlRestApplication.triggerShutdown

Both calls wind up in KsqlEngine.close, and then eventually in KsqlEngine.unregisterQuery. One thread (Thread A) removes the query from allLiveQueries and moves on, and the other thread (lets call it Thread B) sees allLiveQueries doesn't have that query and moves on. Thread A will then go on and try to delete the query's internal topics. However, if Thread B is able to successfully close the admin client before this happens, the topics won't get deleted.

If you introduce a sleep just before the AdminClient is closed, it's pretty reliable to reproduce:

diff --git a/ksql-engine/src/main/java/io/confluent/ksql/services/DefaultServiceContext.java b/ksql-engine/src/main/java/io/confluent/ksql/services/DefaultServiceContext.java
index d7e495d1c..acdb1a00a 100644
--- a/ksql-engine/src/main/java/io/confluent/ksql/services/DefaultServiceContext.java
+++ b/ksql-engine/src/main/java/io/confluent/ksql/services/DefaultServiceContext.java
@@ -140,6 +140,11 @@ public class DefaultServiceContext implements ServiceContext {

   @Override
   public void close() {
+    try {
+      Thread.sleep(30000);
+    } catch (final Exception e) {
+      System.out.println("WE GOT AN ITERRUPT");
+    }
     if (adminClientSupplier.isInitialized()) {
       adminClientSupplier.get().close();
     }

[agavra]

Fixed by #4658