From 8c547ab008f8b3b8b9d325338068a34b302e3bf2 Mon Sep 17 00:00:00 2001
From: miton18 <remi@collignon-ducret.fr>
Date: Wed, 2 Oct 2024 17:57:49 +0200
Subject: [PATCH] Fix pulsar finalize

---
 quickwit/quickwit-indexing/src/source/pulsar_source.rs | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/quickwit/quickwit-indexing/src/source/pulsar_source.rs b/quickwit/quickwit-indexing/src/source/pulsar_source.rs
index 11e65cfe03b..c263a877ef5 100644
--- 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(),