test(stream): add ci for split change mutation in source

[tabVersion]

What's changed and what's your intention?

PLEASE DO NOT LEAVE THIS EMPTY !!!

Please explain IN DETAIL what the changes are in this PR and why they are needed:

add a test for config change in source (split change mutation)

fixing hang behavior if two barriers comes in the first two polls: wrap stream reader in mutex, we may run a bench later

• Summarize your change (mandatory)
• How does this PR work? Need a brief introduction for the changed logic (optional)
• Describe clearly one logical change and avoid lazy messages (optional)
• Describe any limitations of the current code (optional)
• Add the 'user-facing changes' label if your PR contains changes that are visible to users (optional)

Checklist

• I have written necessary docs and comments
• I have added necessary unit tests and integration tests
• All checks passed in ./risedev check (or alias, ./risedev c)

Refer to a related PR or issue link (optional)

close #1285

[skyzh]

Please write all tests in https://github.com/singularity-data/risingwave/blob/main/src/compute/tests/integration_tests.rs, otherwise linker will OOM

[skyzh]

Please write all tests in https://github.com/singularity-data/risingwave/blob/main/src/compute/tests/integration_tests.rs, otherwise linker will OOM

[tabVersion]

Please write all tests in https://github.com/singularity-data/risingwave/blob/main/src/compute/tests/integration_tests.rs, otherwise linker will OOM

will fix

btw the test relies on kafka, how do I configure this when running test

[skyzh]

btw the test relies on kafka, how do I configure this when running test

Unit test should be self-contained and should not depend on external service. Otherwise no developer will be willing to test their own changes with ./risedev test.

[tabVersion]

Right. I will implement a source for testing

[tabVersion]

I will migrate to datagen connector after merge #3124

[codecov]

Codecov Report

Merging #3039 (e0fa53d) into main (c6d1288) will increase coverage by 0.34%.
The diff coverage is 97.80%.

@@            Coverage Diff             @@
##             main    #3039      +/-   ##
==========================================
+ Coverage   73.41%   73.75%   +0.34%     
==========================================
  Files         758      758              
  Lines      103393   103559     +166     
==========================================
+ Hits        75905    76385     +480     
+ Misses      27488    27174     -314     

Flag	Coverage Δ
rust	73.75% <97.80%> (+0.34%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/connector/src/datagen/split.rs	87.50% <ø> (+70.83%)	⬆️
src/connector/src/lib.rs	100.00% <ø> (ø)
src/stream/src/executor/mod.rs	51.22% <ø> (ø)
src/connector/src/datagen/source/reader.rs	85.58% <57.14%> (-1.09%)	⬇️
src/stream/src/executor/source.rs	95.16% <99.42%> (+16.38%)	⬆️
src/source/src/parser/json_parser.rs	99.06% <0.00%> (+0.93%)	⬆️
src/connector/src/base.rs	43.33% <0.00%> (+3.33%)	⬆️
src/connector/src/state.rs	88.88% <0.00%> (+5.55%)	⬆️
src/source/src/lib.rs	88.88% <0.00%> (+11.11%)	⬆️
... and 7 more

📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

[tabVersion]

please just ignore the naming issue. I will write assert later.

[shanicky]

LGTM, we may need to test the performance of the mutex in source.rs later

[skyzh]

It's high probability that this impl will cause deadlock, I guess.

[skyzh]

Please hold this PR a little bit, I need to think for a while...

[skyzh]

LGTM. I have no clear idea on how to make it better. We may just fix it when there's really a deadlock, performance regression, or lose of message. The whole implementation looks counter-intuitive, but there might not be a better solution. Some thoughts below, but might not be applicable.

---

(1) Would be better if being implemented in the following way:

stream_reader is Arc<parking_lot::Mutex<Option<Box<Reader>>>

Before calling .next, we take the option out of the Mutex. After calling next, we put the reader back to Mutex. If mutex is not None, we can use the new reader for the next time.

(2) However, this approach still has problems. How can we ensure the reader is replaced to the new one after we finish process the barrier?

(3) The state machine looks pretty counter-intuitive. I think we should do a rewrite to make everything works. The basic logic should be like:

loop {
  let mut stream = select_with_strategy(stream, barrier);
  #[for_await]
  for item in stream {
    match item {
      Either::Right(normal_barrier) => ...
      Either::Right(conf_change_barrier) => break conf_change_barrier;
      Either::Left(chunk) => yield chunk
    }
  }
  if is_conf_change_barrier {
    change reader;
  }
}

(4) When there's pause and resume barrier in the future, how can we ensure we don't lose data in reader's stream state?