fix: don't use queryId of last terminate command after restore

[vpapavas]

Description

Fixes #6010
The problem is that if the last command before a restart was a terminate, then the first command after the restart gets the same queryId as the terminate. Moreover, if the command topic contains only terminate commands, the next queryId will be 0. To avoid this, we always increase the queryId for every command, whether it has a plan or not

Testing done

Unit tests.

Manual testing

ksql> CREATE STREAM A (COLUMN STRING) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON', partitions=1);

 Message
----------------
 Stream created
----------------
ksql> CREATE STREAM B AS SELECT * FROM A;

 Message
--------------------------------
 Created query with ID CSAS_B_3
--------------------------------
ksql> CREATE STREAM C AS SELECT * FROM A;

 Message
--------------------------------
 Created query with ID CSAS_C_5
--------------------------------
ksql> terminate CSAS_B_3;

 Message
-------------------
 Query terminated.
-------------------
ksql> terminate CSAS_C_5;

 Message
-------------------
 Query terminated.
-------------------

Restart

ksql> CREATE STREAM D AS SELECT * FROM A;

 Message
--------------------------------
 Created query with ID CSAS_D_7
--------------------------------

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[stevenpyzhang]

@vpapavas can you try this test scenario? I'm not sure this is fixing the issue.

Delete any existing command topic
Startup server

create stream test as select * from KSQL_PROCESSING_LOG;
terminate CSAS_TEST_0;
drop stream test;

Stop the server. Start the server up.

create stream test as select level from KSQL_PROCESSING_LOG;

I think even with this change, we'd duplicate the CSAS_TEST_0 as the QueryId

[agavra]

Thanks for the fix Vicky! The test coverage gives me more confidence that this works.

The change, however, I think is fixing the issue with a band-aid instead of fixing it at the root. We're adding more edge cases to the code and making the queryID generation more difficult to reason about. I think we should look for another solution that simplifies queryID generation instead of adding more conditions. Let's chat offline if you need some help coming up with solutions.

[agavra]

LGTM! It looks like you accidentally left some of the previous commit's work (handleRestore). Just for anyone reading this discussion, I have thought about whether this is backwards compatible and i think it is because any old commands that used the old numbering (e.g. the first query was always _0) will have their queryIds persisted. It might break some people's automated tests in the worst case