Fix pulsar finalize

quickwit-oss · Oct 3, 2024 · 8c547ab · 8c547ab
1 parent 060dfd8
commit 8c547ab
Showing 1 changed file with 8 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,14 @@ impl Source for PulsarSource {
         format!("{:?}", self)
     }
 
+    async fn finalize(
+        &mut self,
+        _exit_status: &ActorExitStatus,
+        _ctx: &SourceContext,
+    ) -> anyhow::Result<()> {
+        self.pulsar_consumer.close()
+    }
+
     fn observable_state(&self) -> JsonValue {
         json!({
             "index_id": self.source_runtime.index_id(),