test(sink): add recovery test for sink (risingwavelabs#12701)

zwang28 · Oct 27, 2023 · 92470cb · 92470cb
1 parent bc392aa
commit 92470cb
Show file tree

Hide file tree

Showing 7 changed files with 628 additions and 320 deletions.
diff --git a/ci/scripts/deterministic-it-test.sh b/ci/scripts/deterministic-it-test.sh
@@ -16,6 +16,7 @@ mv target/ci-sim target/sim
 echo "--- Run integration tests in deterministic simulation mode"
 seq $TEST_NUM | parallel MADSIM_TEST_SEED={} NEXTEST_PROFILE=ci-sim \
  cargo nextest run \
+ --no-capture \
  --no-fail-fast \
  --cargo-metadata target/nextest/cargo-metadata.json \
  --binaries-metadata target/nextest/binaries-metadata.json \

diff --git a/src/stream/src/executor/sink.rs b/src/stream/src/executor/sink.rs
@@ -241,7 +241,7 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
                             .await?;
                         if let Some(vnode_bitmap) = barrier.as_update_vnode_bitmap(actor_context.id)
                         {
-                            log_writer.update_vnode_bitmap(vnode_bitmap);
+                            log_writer.update_vnode_bitmap(vnode_bitmap).await?;
                         }
                         yield Message::Barrier(barrier);
                     }
@@ -276,7 +276,7 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
                             .await?;
                         if let Some(vnode_bitmap) = barrier.as_update_vnode_bitmap(actor_context.id)
                         {
-                            log_writer.update_vnode_bitmap(vnode_bitmap);
+                            log_writer.update_vnode_bitmap(vnode_bitmap).await?;
                         }
                         yield Message::Barrier(barrier);
                     }