always increase query id

ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/integration/ClientIntegrationTest.java

@@ -862,7 +862,7 @@ public void shouldListQueries() {
 
     // Then
     assertThat(queries.get(0).getQueryType(), is(QueryType.PERSISTENT));
-    assertThat(queries.get(0).getId(), is("CTAS_" + AGG_TABLE + "_0"));
+    assertThat(queries.get(0).getId(), is("CTAS_" + AGG_TABLE + "_5"));
     assertThat(queries.get(0).getSql(), is(
         "CREATE TABLE " + AGG_TABLE + " WITH (KAFKA_TOPIC='" + AGG_TABLE + "', PARTITIONS=1, REPLICAS=1) AS SELECT\n"
             + "  " + TEST_STREAM + ".STR STR,\n"
@@ -894,7 +894,7 @@ public void shouldDescribeSource() throws Exception {
     assertThat(description.valueFormat(), is("JSON"));
     assertThat(description.readQueries(), hasSize(1));
     assertThat(description.readQueries().get(0).getQueryType(), is(QueryType.PERSISTENT));
-    assertThat(description.readQueries().get(0).getId(), is("CTAS_" + AGG_TABLE + "_0"));
+    assertThat(description.readQueries().get(0).getId(), is("CTAS_" + AGG_TABLE + "_5"));
     assertThat(description.readQueries().get(0).getSql(), is(
         "CREATE TABLE " + AGG_TABLE + " WITH (KAFKA_TOPIC='" + AGG_TABLE + "', PARTITIONS=1, REPLICAS=1) AS SELECT\n"
             + "  " + TEST_STREAM + ".STR STR,\n"

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

@@ -254,15 +254,14 @@ public void processPriorCommands() {
       }
 
       final List<QueuedCommand> compacted = compactor.apply(compatibleCommands);
-      final QueryId greatestQueryId = RestoreCommandsCompactor.greatestQueryId;
       compacted.forEach(
           command -> {
             currentCommandRef.set(new Pair<>(command, clock.instant()));
             RetryUtil.retryWithBackoff(
                 maxRetries,
                 STATEMENT_RETRY_MS,
                 MAX_STATEMENT_RETRY_MS,
-                () -> statementExecutor.handleRestore(command, greatestQueryId),
+                () -> statementExecutor.handleRestore(command),
                 WakeupException.class
             );
             currentCommandRef.set(null);

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

@@ -125,23 +125,19 @@ void handleStatement(final QueuedCommand queuedCommand) {
         queuedCommand.getAndDeserializeCommandId(),
         queuedCommand.getStatus(),
         Mode.EXECUTE,
-        queuedCommand.getOffset(),
-        Optional.empty()
+        queuedCommand.getOffset()
     );
   }
 
-  void handleRestore(final QueuedCommand queuedCommand, final QueryId greatestQueryId) {
+  void handleRestore(final QueuedCommand queuedCommand) {
     throwIfNotConfigured();
 
-    final Optional<QueryId> queryId = greatestQueryId == null
-        ? Optional.empty() : Optional.of(greatestQueryId);
     handleStatementWithTerminatedQueries(
         queuedCommand.getAndDeserializeCommand(commandDeserializer),
         queuedCommand.getAndDeserializeCommandId(),
         queuedCommand.getStatus(),
         Mode.RESTORE,
-        queuedCommand.getOffset(),
-        queryId
+        queuedCommand.getOffset()
     );
   }
 
@@ -195,8 +191,7 @@ private void handleStatementWithTerminatedQueries(
       final CommandId commandId,
       final Optional<CommandStatusFuture> commandStatusFuture,
       final Mode mode,
-      final long offset,
-      final Optional<QueryId> greatestQueryId
+      final long offset
   ) {
     try {
       if (command.getPlan().isPresent()) {
@@ -206,8 +201,7 @@ private void handleStatementWithTerminatedQueries(
             commandStatusFuture,
             command.getPlan().get(),
             mode,
-            offset,
-            greatestQueryId);
+            offset);
         return;
       }
       final String statementString = command.getStatement();
@@ -240,8 +234,7 @@ private void executePlan(
       final Optional<CommandStatusFuture> commandStatusFuture,
       final KsqlPlan plan,
       final Mode mode,
-      final long offset,
-      final Optional<QueryId> greatestQueryId
+      final long offset
   ) {
     final KsqlConfig mergedConfig = buildMergedConfig(command);
     final ConfiguredKsqlPlan configured = ConfiguredKsqlPlan.of(
@@ -254,15 +247,8 @@ private void executePlan(
         new CommandStatus(CommandStatus.Status.EXECUTING, "Executing statement")
     );
     final ExecuteResult result = ksqlEngine.execute(serviceContext, configured);
-    long queryID = Long.MIN_VALUE;
-    if (greatestQueryId.isPresent() && mode == Mode.RESTORE) {
-      final String ltq = greatestQueryId.get().toString();
-      final int lastIndex = ltq.lastIndexOf("_");
-      queryID = Long.parseLong(ltq.substring(lastIndex + 1));
-      queryIdGenerator.setNextId(queryID + 1);
-    }
+    queryIdGenerator.setNextId(offset + 1);
     if (result.getQuery().isPresent()) {
-      queryIdGenerator.setNextId(offset + 1);
       if (mode == Mode.EXECUTE) {
         result.getQuery().get().start();
       }

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

@@ -24,17 +24,12 @@
 import java.util.Map;
 import java.util.Optional;
 import javax.annotation.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Util for compacting the restore commands
  */
 public final class RestoreCommandsCompactor {
 
-  static QueryId greatestQueryId;
-  private static final Logger LOG = LoggerFactory.getLogger(RestoreCommandsCompactor.class);
-
   private RestoreCommandsCompactor() {
   }
 
@@ -91,8 +86,6 @@ public static CompactedNode maybeAppend(
       if (queued.getAndDeserializeCommandId().getType() == Type.TERMINATE) {
         final QueryId queryId = new QueryId(queued.getAndDeserializeCommandId().getEntity());
         markShouldSkip(queryId, latestNodeWithId);
-        //keep track of the last terminate command
-        //lastTerminateQueryId = queryId;
 
         // terminate commands don't get added to the list of commands to execute
         // because we "execute" them in this class by removing query plans from
@@ -135,12 +128,6 @@ private CompactedNode(
   private static Optional<QueuedCommand> compact(final CompactedNode node) {
     final Command command = node.command;
 
-    if (command.getPlan().get().getQueryPlan().isPresent()) {
-      if (greatestQueryId == null) {
-        greatestQueryId = command.getPlan().get().getQueryPlan().get().getQueryId();
-      }
-    }
-
     if (!node.shouldSkip) {
       return Optional.of(node.queued);
     }
@@ -165,9 +152,4 @@ private static Optional<QueuedCommand> compact(final CompactedNode node) {
         node.queued.getOffset()
     ));
   }
-
-  static QueryId getGreatestQueryId() {
-    return greatestQueryId;
-  }
-
 }

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

@@ -960,7 +960,7 @@ private void handleStatement(
 
   private void terminateQueries() {
     final Command terminateCommand1 = new Command(
-        "TERMINATE CSAS_USER1PV_0;",
+        "TERMINATE CSAS_USER1PV_1;",
         emptyMap(),
         ksqlConfig.getAllConfigPropsWithSecretsObfuscated(),
         Optional.empty()

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

@@ -580,7 +580,7 @@ public void shouldRecoverRecreates() {
     server1.submitCommands(
         "CREATE STREAM A (ROWKEY STRING KEY, C1 STRING, C2 INT) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT ROWKEY, C1 FROM A;",
-        "TERMINATE CsAs_b_0;",
+        "TERMINATE CsAs_b_1;",
         "DROP STREAM B;",
         "CREATE STREAM B AS SELECT ROWKEY, C2 FROM A;"
     );
@@ -603,7 +603,7 @@ public void shouldRecoverReplacesWithTerminates() {
         "CREATE STREAM A (ROWKEY STRING KEY, C1 STRING, C2 INT) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT ROWKEY, C1 FROM A;",
         "CREATE OR REPLACE STREAM B AS SELECT ROWKEY, C1, C2 FROM A;",
-        "TERMINATE CSAS_B_0;",
+        "TERMINATE CSAS_B_1;",
         "DROP STREAM B;",
         "CREATE STREAM B AS SELECT ROWKEY, C1 FROM A;"
     );
@@ -627,7 +627,7 @@ public void shouldRecoverInsertIntosRecreates() {
         "CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B (COLUMN STRING) WITH (KAFKA_TOPIC='B', VALUE_FORMAT='JSON', PARTITIONS=1);",
         "INSERT INTO B SELECT * FROM A;",
-        "TERMINATE InsertQuery_0;",
+        "TERMINATE InsertQuery_2;",
         "INSERT INTO B SELECT * FROM A;"
     );
     shouldRecover(commands);
@@ -639,7 +639,7 @@ public void shouldRecoverTerminates() {
         "CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT * FROM A;",
         "INSERT INTO B SELECT * FROM A;",
-        "TERMINATE CSAS_B_0;",
+        "TERMINATE CSAS_B_1;",
         "TERMINATE InsertQuery_2;"
     );
     shouldRecover(commands);
@@ -650,7 +650,7 @@ public void shouldRecoverDrop() {
     server1.submitCommands(
         "CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT * FROM A;",
-        "TERMINATE CSAS_B_0;",
+        "TERMINATE CSAS_B_1;",
         "DROP STREAM B;"
     );
     shouldRecover(commands);
@@ -661,7 +661,7 @@ public void shouldRecoverWithDuplicateTerminateAndDrop() {
     server1.submitCommands(
         "CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT * FROM A;",
-        "TERMINATE CSAS_B_0;"
+        "TERMINATE CSAS_B_1;"
     );
 
     addDuplicateOfLastCommand(); // Add duplicate of "TERMINATE CSAS_B_0;"
@@ -681,7 +681,7 @@ public void shouldNotDeleteTopicsOnRecovery() {
     server1.submitCommands(
         "CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT * FROM A;",
-        "TERMINATE CSAS_B_0;",
+        "TERMINATE CSAS_B_1;",
         "DROP STREAM B DELETE TOPIC;"
     );
 
@@ -704,39 +704,39 @@ public void shouldRecoverQueryIDs() {
     final Set<QueryId> queryIdNames = queriesById(server.ksqlEngine.getPersistentQueries())
         .keySet();
 
-    assertThat(queryIdNames, contains(new QueryId("CSAS_C_0")));
+    assertThat(queryIdNames, contains(new QueryId("CSAS_C_1")));
   }
 
   @Test
   public void shouldIncrementQueryIDsNoPlans() {
     server1.submitCommands(
         "CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT * FROM A;",
-        "TERMINATE CSAS_B_0;");
+        "TERMINATE CSAS_B_1;");
 
     final KsqlServer server = new KsqlServer(commands);
     server.recover();
     server.submitCommands("CREATE STREAM C AS SELECT * FROM A;");
     final Set<QueryId> queryIdNames = queriesById(server.ksqlEngine.getPersistentQueries())
         .keySet();
 
-    assertThat(queryIdNames, contains(new QueryId("CSAS_C_1")));
+    assertThat(queryIdNames, contains(new QueryId("CSAS_C_2")));
   }
 
   @Test
   public void shouldIncrementQueryIDsWithPlan() {
     server1.submitCommands(
         "CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT * FROM A;",
-        "TERMINATE CSAS_B_0;");
+        "TERMINATE CSAS_B_1;");
 
     final KsqlServer server = new KsqlServer(commands);
     server.recover();
     server.submitCommands("CREATE STREAM C AS SELECT * FROM A;");
     final Set<QueryId> queryIdNames = queriesById(server.ksqlEngine.getPersistentQueries())
         .keySet();
 
-    assertThat(queryIdNames, contains(new QueryId("CSAS_C_1")));
+    assertThat(queryIdNames, contains(new QueryId("CSAS_C_2")));
   }