test: recovery test for sources

[tabVersion]

• Background

In our previous tests, we did not thoroughly validate whether each source connector could resume consumption from any specific record position while ensuring exactly-once processing. The primary challenges in achieving this were the need to restart clusters within the CI environment to perform recovery operations and the inability to control the consumption rate to manage progress.

• Motivation

The absence of specific tests led to unintentional breaches of the exactly-once semantics when building new sources. This has been a critical issue, especially since we identified that the fs source connector could duplicate reading the current message during recovery, thanks to @stdrc . By implementing these tests, we aim to strengthen the guarantees around exactly-once semantics across all connectors.

• Prerequisites

We now support triggering recovery via SQL commands (#16259), controlling read speeds with rate limits (#15948), and supporting truncation at any position for chunks. Our new support for bash commands in the slt (#12451 (comment)) makes it easier to control external components. The inclusion of 'key as' syntax allows clear marking of each message's offset to check for overlaps (#13707).

• Procedure

To test recovery, we can use a smaller dataset, for example, 20 messages in Kafka, setting stream parallelism to 1 and streaming_rate_limit to 1. We will trigger recovery at any time between 0-20 seconds to ensure that all 20 messages are read correctly and to verify that there are no duplicates in the offsets. The same test applies to the fs source; if a data line is read twice, the recorded offset will be the same.

This test will help ensure that all our connectors meet the exactly-once requirements, safeguarding the integrity of our data processing systems.

[kwannoel]

Agree! Besides e2e unit tests, I also would like to have fuzzing similar to that of e2e deterministic recovery test.

[kwannoel]

We can use inline system commands like ./risedev k and ./risedev d to start and stop the cluster inline.
After running the recovery part, sleep for some duration, and check that the records should have been ingested still.

Test Variables:

1. DDL: create sink, create source + mv, create table with source.
2. recovery: crash loop scenario (keep triggering restarts), long recovery (20mins).

[xxchan]

I've managed to using the RECOVER command to write recovery tests https://github.com/risingwavelabs/risingwave/pull/16733/files#diff-b83fa16ce469bbf54f92cf95f9d778e4cd0acf6a20489589c8bb636b6162da02

[xxchan]

We can use inline system commands like ./risedev k and ./risedev d to start and stop the cluster inline.

To do this, sqllogictest need to be enhanced, because the connection to fe will be disconnected (#12451 (comment))

[tabVersion]

We can use inline system commands like ./risedev k and ./risedev d to start and stop the cluster inline.

To do this, sqllogictest need to be enhanced, because the connection to fe will be disconnected (#12451 (comment))

IIRC, the recovery process just happens in meta inner, any connection to the fe will not be influenced.

[st1page]

Motivation
The absence of specific tests led to unintentional breaches of the exactly-once semantics when building new sources. This has been a critical issue, especially since we identified that the fs source connector could duplicate reading the current message during recovery, thanks to @stdrc . By implementing these tests, we aim to strengthen the guarantees around exactly-once semantics across all connectors.

In addition to the exactly-once semantic guarantee, it is also meaningful to simply test whether re-creating the subscription to the upstream system after recovery will cause problems, such as the issue in #17112