From c141addf9ed268ebe952b930aea21210e028013b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Collignon-Ducret=20R=C3=A9mi?=
 <miton18@users.noreply.github.com>
Date: Wed, 9 Oct 2024 21:07:47 +0200
Subject: [PATCH] Fix pulsar finalize (#5471)

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

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