fix: idempotent terminate that can handle hung streams (#4643)

Fixes a couple issues with terminate:
- don't throw if the query doesn't get into NOT_RUNNING state. This can
  happen when streams threads are stuck pending shutdown.
- make terminate idempotent

ksql-engine/src/main/java/io/confluent/ksql/engine/KsqlEngine.java

@@ -264,8 +264,10 @@ private void unregisterQuery(final ServiceContext serviceContext, final QueryMet
     final String applicationId = query.getQueryApplicationId();
 
     if (!query.getState().equalsIgnoreCase("NOT_RUNNING")) {
-      throw new IllegalStateException("query not stopped."
-          + " id " + applicationId + ", state: " + query.getState());
+      log.warn(
+          "Unregistering query that has not terminated. "
+              + "This may happen when streams threads are hung. State: " + query.getState()
+      );
     }
 
     if (!allLiveQueries.remove(query)) {

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/InteractiveStatementExecutor.java

@@ -359,10 +359,8 @@ private void terminateQuery(final PreparedStatement<TerminateQuery> terminateQue
       return;
     }
 
-    ksqlEngine.getPersistentQuery(queryId.get())
-        .orElseThrow(() ->
-            new KsqlException(String.format("No running query with id %s was found", queryId)))
-        .close();
+    final Optional<PersistentQueryMetadata> query = ksqlEngine.getPersistentQuery(queryId.get());
+    query.ifPresent(PersistentQueryMetadata::close);
   }
 
 }

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/InteractiveStatementExecutorTest.java

@@ -22,6 +22,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.argThat;
@@ -715,6 +716,37 @@ public void shouldTerminateAll() {
     verify(query1).close();
   }
 
+  @Test
+  public void shouldDoIdempotentTerminate() {
+    // Given:
+    final String queryStatement = "a persistent query";
+
+    final TerminateQuery terminate = mock(TerminateQuery.class);
+    when(terminate.getQueryId()).thenReturn(Optional.of(new QueryId("foo")));
+
+    when(mockParser.parseSingleStatement(any()))
+        .thenReturn(PreparedStatement.of(queryStatement, terminate));
+
+    final PersistentQueryMetadata query = mock(PersistentQueryMetadata.class);
+
+    when(mockEngine.getPersistentQuery(new QueryId("foo")))
+        .thenReturn(Optional.of(query))
+        .thenReturn(Optional.empty());
+
+    final QueuedCommand cmd = new QueuedCommand(
+        new CommandId(Type.TERMINATE, "-", Action.EXECUTE),
+        new Command("terminate all", emptyMap(), emptyMap(), Optional.empty()),
+        Optional.empty(),
+        0L
+    );
+
+    // When:
+    statementExecutorWithMocks.handleStatement(cmd);
+    statementExecutorWithMocks.handleStatement(cmd);
+
+    // Then should not throw
+  }
+
   private void createStreamsAndStartTwoPersistentQueries() {
     final Command csCommand = new Command(
         "CREATE STREAM pageview ("