chore(deps): Bump pulsar-rs to latest SHA

This pulls in a fix for the connection_manager prematurely deleting
connections that can be used for `update_topics` (which runs
automatically in the background). Pulsar has not released an official
crate for this yet, so we'll use the commit SHA.

Ref: LOG-23224

Author: darinspivey

Cargo.lock

@@ -386,7 +386,7 @@ openssl-probe = { version = "0.1.6", default-features = false }
 ordered-float = { version = "4.6.0", default-features = false, features = ["serde"] }
 percent-encoding = { version = "2.3.1", default-features = false }
 postgres-openssl = { version = "0.5.1", default-features = false, features = ["runtime"], optional = true }
-pulsar = { git = "https://github.com/streamnative/pulsar-rs", rev = "5dfd10ce1c92d3200eb9cc010ab224ded46c9adc", default-features = false, features = ["tokio-runtime", "auth-oauth2", "flate2", "lz4", "snap", "zstd"], optional = true }
+pulsar = { git = "https://github.com/streamnative/pulsar-rs", rev = "174070149bab64359f8a31094d7bfb857b045bf1", default-features = false, features = ["tokio-runtime", "auth-oauth2", "flate2", "lz4", "snap", "zstd"], optional = true }
 quick-junit = { version = "0.5.1" }
 rand.workspace = true
 rand_distr.workspace = true

Cargo.toml

@@ -308,6 +308,7 @@ impl PulsarSinkConfig {
                             .map_or(default_retry_options.keep_alive, |secs| {
                                 Duration::from_secs(secs)
                             }),
+                        ..ConnectionRetryOptions::default()
                     }
                 });
 

src/sinks/pulsar/config.rs

@@ -329,6 +329,10 @@ pub struct PulsarConnectionRetryOptions {
     #[configurable(metadata(docs::type_unit = "seconds"))]
     #[configurable(metadata(docs::examples = 60))]
     pub keep_alive_secs: Option<u64>,
+    /// Maximum idle time before a connection is eligible for cleanup
+    #[configurable(metadata(docs::type_unit = "seconds"))]
+    #[configurable(metadata(docs::examples = 120))]
+    pub connection_max_idle: Option<u64>,
 }
 
 #[derive(Debug)]
@@ -436,6 +440,9 @@ impl PulsarSourceConfig {
             if let Some(secs) = opts.keep_alive_secs {
                 retry_options.keep_alive = Duration::from_secs(secs);
             }
+            if let Some(secs) = opts.connection_max_idle {
+                retry_options.connection_max_idle = Duration::from_secs(secs);
+            }
         }
 
         debug!("Creating pulsar consumer with options: {:?}", retry_options);
@@ -871,6 +878,7 @@ mod tests {
             max_retries = 50
             connection_timeout_secs = 59
             keep_alive_secs = 58
+            connection_max_idle = 55
         "#,
         );
 
@@ -880,6 +888,7 @@ mod tests {
         assert_eq!(opts.max_retries, Some(50));
         assert_eq!(opts.connection_timeout_secs, Some(59));
         assert_eq!(opts.keep_alive_secs, Some(58));
+        assert_eq!(opts.connection_max_idle, Some(55));
     }
 
     #[test]
@@ -926,6 +935,10 @@ mod tests {
             opts.keep_alive_secs, None,
             "Expected keep_alive_secs to be None"
         );
+        assert_eq!(
+            opts.connection_max_idle, None,
+            "Expected connection_max_idle to be None"
+        );
     }
 
     #[test]