flush kafka on close (#619)

ArroyoSystems · May 8, 2024 · 7e69de7 · 7e69de7
1 parent 8e5e26c
commit 7e69de7
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/crates/arroyo-connectors/src/kafka/sink/mod.rs b/crates/arroyo-connectors/src/kafka/sink/mod.rs
@@ -242,6 +242,7 @@ impl ArrowOperator for KafkaSinkFunc {
     }
 
     async fn on_close(&mut self, _: &Option<SignalMessage>, ctx: &mut ArrowContext) {
+        self.flush(ctx).await;
         if !self.is_committing() {
             return;
         }