still call flush_current_epoch on barrier after pause

risingwavelabs · Feb 26, 2024 · 3689bc6 · 3689bc6
1 parent 9969c10
commit 3689bc6
Showing 1 changed file with 3 additions and 6 deletions.
diff --git a/src/stream/src/executor/sink.rs b/src/stream/src/executor/sink.rs
@@ -216,12 +216,9 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
                     yield Message::Chunk(chunk);
                 }
                 Message::Barrier(barrier) => {
-                    // No need to flush barrier to log writer when paused because there will be no data between pause barrier and resume barrier.
-                    if !is_paused {
-                        log_writer
-                            .flush_current_epoch(barrier.epoch.curr, barrier.kind.is_checkpoint())
-                            .await?;
-                    }
+                    log_writer
+                        .flush_current_epoch(barrier.epoch.curr, barrier.kind.is_checkpoint())
+                        .await?;
 
                     if let Some(mutation) = barrier.mutation.as_deref() {
                         match mutation {