fix: wrap fetch_watermarks in spawn_blocking

CHANGELOG.md

@@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## rdkafka [0.3.3] - 2024-02-28
+
+### Changed
+
+- Wrap `fetch_watermarks` in `tokio::task::spawn_blocking`.
+
 ## rdkafka [0.3.2] - 2024-02-28
 
 ### Changed

madsim-rdkafka/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "madsim-rdkafka"
-version = "0.3.2+0.34.0"
+version = "0.3.3+0.34.0"
 edition = "2021"
 authors = ["Runji Wang <[email protected]>"]
 description = "The rdkafka simulator on madsim."

madsim-rdkafka/README.md

@@ -25,7 +25,7 @@ The following functions are modified to be `async`:
 - `ClientConfig::create`
 - `ClientConfig::create_with_context`
 - `Client::fetch_metadata`
-- `Client::fetch_watermarks`
+- `Client::fetch_watermarks`[^1]
 - `Client::fetch_group_list`
 - `Consumer::seek`
 - `Consumer::seek_partitions`
@@ -37,14 +37,16 @@ The following functions are modified to be `async`:
 - `Consumer::offsets_for_timestamp`
 - `Consumer::offsets_for_times`
 - `Consumer::fetch_metadata`
-- `Consumer::fetch_watermarks`
+- `Consumer::fetch_watermarks`[^1]
 - `Consumer::fetch_group_list`
 - `Producer::flush`
 - `Producer::init_transactions`
 - `Producer::send_offsets_to_transaction`
 - `Producer::commit_transaction`
 - `Producer::abort_transaction`
 
+[^1]: wrapped in `tokio::task::spawn_blocking`
+
 ## DNS Resolution
 
 This crate has cherry-picked [a commit] from Materialize to support rewriting broker addresses.

madsim-rdkafka/src/std/client.rs

@@ -354,29 +354,32 @@ impl<C: ClientContext> Client<C> {
     }
 
     /// Returns high and low watermark for the specified topic and partition.
-    pub async fn fetch_watermarks<T: Into<Timeout>>(
+    pub async fn fetch_watermarks<T: Into<Timeout> + Send + 'static>(
         &self,
         topic: &str,
         partition: i32,
         timeout: T,
     ) -> KafkaResult<(i64, i64)> {
-        let mut low = -1;
-        let mut high = -1;
         let topic_c = CString::new(topic.to_string())?;
-        let ret = unsafe {
-            rdsys::rd_kafka_query_watermark_offsets(
-                self.native_ptr(),
+        let native_client = unsafe { NativeClient::from_ptr(self.native_ptr()) };
+        tokio::task::spawn_blocking(move || unsafe {
+            let mut low = -1;
+            let mut high = -1;
+            let ret = rdsys::rd_kafka_query_watermark_offsets(
+                native_client.ptr(),
                 topic_c.as_ptr(),
                 partition,
                 &mut low as *mut i64,
                 &mut high as *mut i64,
                 timeout.into().as_millis(),
-            )
-        };
-        if ret.is_error() {
-            return Err(KafkaError::MetadataFetch(ret.into()));
-        }
-        Ok((low, high))
+            );
+            if ret.is_error() {
+                return Err(KafkaError::MetadataFetch(ret.into()));
+            }
+            Ok((low, high))
+        })
+        .await
+        .unwrap()
     }
 
     /// Returns the cluster identifier option or None if the cluster identifier is null

madsim-rdkafka/src/std/consumer/base_consumer.rs

@@ -578,7 +578,7 @@ where
         self.client.fetch_metadata(topic, timeout).await
     }
 
-    async fn fetch_watermarks<T: Into<Timeout> + Send>(
+    async fn fetch_watermarks<T: Into<Timeout> + Send + 'static>(
         &self,
         topic: &str,
         partition: i32,

madsim-rdkafka/src/std/consumer/mod.rs

@@ -391,7 +391,7 @@ where
         timeout: T,
     ) -> KafkaResult<(i64, i64)>
     where
-        T: Into<Timeout> + Send,
+        T: Into<Timeout> + Send + 'static,
         Self: Sized;
 
     /// Returns the group membership information for the given group. If no group is

madsim-rdkafka/src/std/consumer/stream_consumer.rs

@@ -526,7 +526,7 @@ where
         timeout: T,
     ) -> KafkaResult<(i64, i64)>
     where
-        T: Into<Timeout> + Send,
+        T: Into<Timeout> + Send + 'static,
         Self: Sized,
     {
         self.base.fetch_watermarks(topic, partition, timeout).await