feat: don't start queries when corruption is detected during startup

[stevenpyzhang]

Description

If ksqlDB detects corruption during a restore it shouldn't start queries, because the queries may be terminated by a command following the offset at which we detect corruption. Instead, we should only apply DDLs and all queries should be reported as "failing/error" from the API.

Testing done

Unit test

Also, started up a server with backup enables, and created 3 queries. Stopped the server and modified the backup file so that the last query created was different from what's in the command topic. Started up server again and verified that the corruption log was present and the show queries reported the 2 pre-corruption queries in ERROR state.

ksql> show queries;

 Query ID     | Query Type | Status  | Sink Name | Sink Kafka Topic | Query String                                                                                                                                  
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 CSAS_TEST_1  | PERSISTENT | ERROR:1 | TEST      | TEST             | CREATE STREAM TEST WITH (KAFKA_TOPIC='TEST', PARTITIONS=1, REPLICAS=1) AS SELECT * FROM KSQL_PROCESSING_LOG KSQL_PROCESSING_LOG EMIT CHANGES; 
 CSAS_TEST2_3 | PERSISTENT | ERROR:1 | TEST2     | TEST2            | CREATE STREAM TEST2 WITH (KAFKA_TOPIC='TEST2', PARTITIONS=1, REPLICAS=1) AS SELECT * FROM TEST TEST EMIT CHANGES;                             
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------
For detailed information on a Query run: EXPLAIN <Query ID>;
WARNING: The server has detected corruption in the command topic due to modifications performed on it. 
DDL statements will not be processed any further.
Queries may not run properly while the server is in a corrupted state.
If a backup of the command topic is available, restore the command topic using the backup file.
A server restart is required to restore full functionality.

ksql> show queries extended;
ID                   : CSAS_TEST2_3
Query Type           : PERSISTENT
SQL                  : CREATE STREAM TEST2 WITH (KAFKA_TOPIC='TEST2', PARTITIONS=1, REPLICAS=1) AS SELECT *
FROM TEST TEST
EMIT CHANGES;
Host Query Status    : {stevenz-mbp13.lan:8088=ERROR}
...
...
Error Date           : 2021-07-21 11:10:38,522 (PDT)
Error Details        : Query not started due to corruption in the command topic.
Error Type           : USER

WARNING: The server has detected corruption in the command topic due to modifications performed on it. 
DDL statements will not be processed any further.
Queries may not run properly while the server is in a corrupted state.
If a backup of the command topic is available, restore the command topic using the backup file.
A server restart is required to restore full functionality.

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 #")

[wcarlson5]

LGTM. I am not sure the Create state always needs to be mapped to error but I think it should work.

[guozhangwang]

I have a small concern about mapping CREATED to ERROR: the transition of CREATED to REBALANCING is timing dependent, i.e. it would transit after it reached the broker (group coordinator) for the first time since starting up. In practice it is usually very fast, maybe in a few ms, but still before that the CREATED state is actually valid.

For list-queries / query-status purposes, if the broker was temporarily unavailable after the KS runtime is up and running, then the state may stuck in CREATED for a while, causing them to report false positives.

Some background context about this on JIRA: https://issues.apache.org/jira/browse/KAFKA-6520?src=confmacro

[stevenpyzhang]

@guozhangwang @wcarlson5 I updated the PR so we're not updating the Streams state to Query state mapping. Instead, I just set a corruptionDetection variable in QueryMetadataImpl and the QueryMetadata.getState() returns KafkaStreams ERROR state if the corruption variable is set.

[guozhangwang]

LGTM!

[wcarlson5]

Good idea