Fix pulsar finalize (#5471)

quickwit-oss · Oct 9, 2024 · c141add · c141add
1 parent 6d3011d
commit c141add
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/quickwit/quickwit-indexing/src/source/pulsar_source.rs b/quickwit/quickwit-indexing/src/source/pulsar_source.rs
@@ -274,6 +274,15 @@ impl Source for PulsarSource {
         format!("{:?}", self)
     }
 
+    async fn finalize(
+        &mut self,
+        _exit_status: &ActorExitStatus,
+        _ctx: &SourceContext,
+    ) -> anyhow::Result<()> {
+        self.pulsar_consumer.close().await?;
+        Ok(())
+    }
+
     fn observable_state(&self) -> JsonValue {
         json!({
             "index_id": self.source_runtime.index_id(),