Update from upstream c6d9a65a6df8c43fdf93b42bb276b6b7d2ba6c23

.github/workflows/ci.yml

@@ -7,19 +7,20 @@ on:
     branches: [master]
 
 env:
-  rust_version: 1.61.0
+  rust_version: 1.70.0
 
 jobs:
   lint:
     runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v3
-      - uses: dtolnay/rust-toolchain@b44cb146d03e8d870c57ab64b80f04586349ca5d
+      - uses: dtolnay/rust-toolchain@stable
         with:
           toolchain: ${{ env.rust_version }}
           components: rustfmt, clippy
       - run: cargo fmt -- --check
       - run: cargo clippy -- -Dwarnings
+      - run: cargo clippy --tests -- -Dwarnings
       - run: cargo test --doc
 
   check:
@@ -38,7 +39,8 @@ jobs:
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v3
-      - uses: dtolnay/rust-toolchain@b44cb146d03e8d870c57ab64b80f04586349ca5d
+      - uses: lukka/get-cmake@latest
+      - uses: dtolnay/rust-toolchain@stable
         with:
           toolchain: ${{ env.rust_version }}
       - run: cargo build --all-targets --verbose --features "${{ matrix.features }}"
@@ -50,41 +52,44 @@ jobs:
     runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v3
-      - uses: dtolnay/rust-toolchain@b44cb146d03e8d870c57ab64b80f04586349ca5d
+      - uses: dtolnay/rust-toolchain@stable
         with:
           # The version of this toolchain doesn't matter much. It's only used to
           # generate the minimal-versions lockfile, not to actually run `cargo
           # check`.
           toolchain: nightly
           components: rustfmt, clippy
-      - uses: dtolnay/rust-toolchain@b44cb146d03e8d870c57ab64b80f04586349ca5d
+      - uses: dtolnay/rust-toolchain@stable
         with:
           toolchain: ${{ env.rust_version }}
       - run: rustup default ${{ env.rust_version }}
       - run: cargo +nightly -Z minimal-versions generate-lockfile
-      - run: cargo check
+      # Default features and features that require optional dependencies should be
+      # explicitly checked.
+      - run: cargo check --features libz,tokio,tracing
 
   test:
     strategy:
       fail-fast: false
+      # The test suite doesn't support concurrent runs.
+      max-parallel: 1
       matrix:
         include:
+          - confluent-version: 7.7.0
+            kafka-version: 3.7
+          - confluent-version: 7.5.1
+            kafka-version: 3.6
           - confluent-version: 7.5.1
             kafka-version: 3.5
-          - confluent-version: 5.3.1
-            kafka-version: 2.3
-          - confluent-version: 5.0.3
-            kafka-version: 2.0
-          - confluent-version: 4.1.3
-            kafka-version: 1.1
     runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v3
-      - uses: dtolnay/rust-toolchain@b44cb146d03e8d870c57ab64b80f04586349ca5d
+      - uses: lukka/get-cmake@latest
+      - uses: dtolnay/rust-toolchain@stable
         with:
           toolchain: ${{ env.rust_version }}
-      - run: sudo apt-get update
-      - run: sudo apt-get install -qy valgrind
+      # - run: sudo apt-get update
+      # - run: sudo apt-get install -qy valgrind  # Valgrind currently disabled in testing
       - run: ./test_suite.sh
         env:
           CONFLUENT_VERSION: ${{ matrix.confluent-version }}

Cargo.toml

@@ -10,7 +10,7 @@ keywords = ["kafka", "rdkafka"]
 categories = ["api-bindings"]
 edition = "2018"
 exclude = ["Cargo.lock"]
-rust-version = "1.61"
+rust-version = "1.70"
 
 [workspace]
 members = ["rdkafka-sys"]

README.md

@@ -184,7 +184,7 @@ re-exported as rdkafka features.
 
 ### Minimum supported Rust version (MSRV)
 
-The current minimum supported Rust version (MSRV) is 1.61.0. Note that
+The current minimum supported Rust version (MSRV) is 1.70.0. Note that
 bumping the MSRV is not considered a breaking change. Any release of
 rust-rdkafka may bump the MSRV.
 

changelog.md

@@ -4,6 +4,10 @@ See also the [rdkafka-sys changelog](rdkafka-sys/changelog.md).
 
 ## Unreleased
 
+* Update MSRV to 1.70
+* Remove testign for old Kafka versions (before 3.0). Add tests for 3.7.
+* Fix test dependency on docker compose.
+
 ## 0.36.2 (2024-01-16)
 
 * Update `BaseConsumer::poll` to return `None` when handling rebalance

docker-compose.yaml

@@ -1,5 +1,3 @@
-version: '3'
-
 services:
   kafka:
     image: confluentinc/cp-kafka:${CONFLUENT_VERSION:-7.5.1}

rdkafka-sys/Cargo.toml

@@ -10,7 +10,7 @@ description = "Native bindings to the librdkafka library"
 keywords = ["kafka", "rdkafka"]
 categories = ["external-ffi-bindings"]
 edition = "2018"
-rust-version = "1.61"
+rust-version = "1.70"
 
 [dependencies]
 num_enum = "0.5.0"

rdkafka-sys/build.rs

@@ -77,7 +77,7 @@ fn main() {
         // Ensure that we are in the right directory
         let rdkafkasys_root = Path::new("rdkafka-sys");
         if rdkafkasys_root.exists() {
-            assert!(env::set_current_dir(&rdkafkasys_root).is_ok());
+            assert!(env::set_current_dir(rdkafkasys_root).is_ok());
         }
         if !Path::new("librdkafka/LICENSE").exists() {
             eprintln!("Setting up submodules");

rdkafka-sys/src/lib.rs

@@ -100,7 +100,7 @@ extern crate sasl2_sys;
 #[cfg(feature = "libz-sys")]
 extern crate libz_sys;
 
-#[cfg(any(feature = "curl-sys", feature = "curl-sys/static-curl"))]
+#[cfg(any(feature = "curl-sys", feature = "curl-static"))]
 extern crate curl_sys;
 
 #[cfg(feature = "zstd-sys")]

rdkafka-sys/src/types.rs

@@ -78,6 +78,9 @@ pub type RDKafkaDeleteGroup = bindings::rd_kafka_DeleteGroup_t;
 /// Native rdkafka new partitions object.
 pub type RDKafkaNewPartitions = bindings::rd_kafka_NewPartitions_t;
 
+/// Native rdkafka delete records object.
+pub type RDKafkaDeleteRecords = bindings::rd_kafka_DeleteRecords_t;
+
 /// Native rdkafka config resource.
 pub type RDKafkaConfigResource = bindings::rd_kafka_ConfigResource_t;
 

src/admin.rs

@@ -26,6 +26,7 @@ use crate::config::{ClientConfig, FromClientConfig, FromClientConfigAndContext};
 use crate::error::{IsError, KafkaError, KafkaResult};
 use crate::log::{trace, warn};
 use crate::util::{cstr_to_owned, AsCArray, ErrBuf, IntoOpaque, KafkaDrop, NativePtr, Timeout};
+use crate::TopicPartitionList;
 
 //
 // ********** ADMIN CLIENT **********
@@ -218,6 +219,53 @@ impl<C: ClientContext> AdminClient<C> {
         Ok(rx)
     }
 
+    /// Deletes records from a topic.
+    ///
+    /// The provided `offsets` is a topic partition list specifying which
+    /// records to delete from a list of topic partitions. For each entry in the
+    /// list, the messages at offsets before the specified offsets (exclusive)
+    /// in the specified partition will be deleted. Use offset [`Offset::End`]
+    /// to delete all records in the partition.
+    ///
+    /// Returns a topic partition list describing the result of the deletion. If
+    /// the operation succeeded for a partition, the offset for that partition
+    /// will be set to the post-deletion low-water mark for that partition. If
+    /// the operation failed for a partition, there will be an error for that
+    /// partition's entry in the list.
+    pub fn delete_records(
+        &self,
+        offsets: &TopicPartitionList,
+        opts: &AdminOptions,
+    ) -> impl Future<Output = KafkaResult<TopicPartitionList>> {
+        match self.delete_records_inner(offsets, opts) {
+            Ok(rx) => Either::Left(DeleteRecordsFuture { rx }),
+            Err(err) => Either::Right(future::err(err)),
+        }
+    }
+
+    fn delete_records_inner(
+        &self,
+        offsets: &TopicPartitionList,
+        opts: &AdminOptions,
+    ) -> KafkaResult<oneshot::Receiver<NativeEvent>> {
+        let mut err_buf = ErrBuf::new();
+        let delete_records = unsafe {
+            NativeDeleteRecords::from_ptr(rdsys::rd_kafka_DeleteRecords_new(offsets.ptr()))
+        }
+        .ok_or_else(|| KafkaError::AdminOpCreation(err_buf.to_string()))?;
+        let (native_opts, rx) = opts.to_native(self.client.native_ptr(), &mut err_buf)?;
+        unsafe {
+            rdsys::rd_kafka_DeleteRecords(
+                self.client.native_ptr(),
+                &mut delete_records.ptr(),
+                1,
+                native_opts.ptr(),
+                self.queue.ptr(),
+            );
+        }
+        Ok(rx)
+    }
+
     /// Retrieves the configuration parameters for the specified resources.
     ///
     /// Note that while the API supports describing multiple configurations at
@@ -950,6 +998,43 @@ impl Future for CreatePartitionsFuture {
     }
 }
 
+//
+// Delete records handling
+//
+
+type NativeDeleteRecords = NativePtr<RDKafkaDeleteRecords>;
+
+unsafe impl KafkaDrop for RDKafkaDeleteRecords {
+    const TYPE: &'static str = "delete records";
+    const DROP: unsafe extern "C" fn(*mut Self) = rdsys::rd_kafka_DeleteRecords_destroy;
+}
+
+struct DeleteRecordsFuture {
+    rx: oneshot::Receiver<NativeEvent>,
+}
+
+impl Future for DeleteRecordsFuture {
+    type Output = KafkaResult<TopicPartitionList>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let event = ready!(self.rx.poll_unpin(cx)).map_err(|_| KafkaError::Canceled)?;
+        event.check_error()?;
+        let res = unsafe { rdsys::rd_kafka_event_DeleteRecords_result(event.ptr()) };
+        if res.is_null() {
+            let typ = unsafe { rdsys::rd_kafka_event_type(event.ptr()) };
+            return Poll::Ready(Err(KafkaError::AdminOpCreation(format!(
+                "delete records request received response of incorrect type ({})",
+                typ
+            ))));
+        }
+        let tpl = unsafe {
+            let tpl = rdsys::rd_kafka_DeleteRecords_result_offsets(res);
+            TopicPartitionList::from_ptr(rdsys::rd_kafka_topic_partition_list_copy(tpl))
+        };
+        Poll::Ready(Ok(tpl))
+    }
+}
+
 //
 // Describe configs handling
 //

src/client.rs

@@ -198,6 +198,21 @@ impl NativeClient {
     }
 }
 
+pub(crate) enum EventPollResult<T> {
+    None,
+    EventConsumed,
+    Event(T),
+}
+
+impl<T> From<EventPollResult<T>> for Option<T> {
+    fn from(val: EventPollResult<T>) -> Self {
+        match val {
+            EventPollResult::None | EventPollResult::EventConsumed => None,
+            EventPollResult::Event(evt) => Some(evt),
+        }
+    }
+}
+
 /// A low-level rdkafka client.
 ///
 /// This type is the basis of the consumers and producers in the [`consumer`]
@@ -278,31 +293,42 @@ impl<C: ClientContext> Client<C> {
         &self.context
     }
 
-    pub(crate) fn poll_event(&self, queue: &NativeQueue, timeout: Timeout) -> Option<NativeEvent> {
+    pub(crate) fn poll_event(
+        &self,
+        queue: &NativeQueue,
+        timeout: Timeout,
+    ) -> EventPollResult<NativeEvent> {
         let event = unsafe { NativeEvent::from_ptr(queue.poll(timeout)) };
         if let Some(ev) = event {
             let evtype = unsafe { rdsys::rd_kafka_event_type(ev.ptr()) };
             match evtype {
-                rdsys::RD_KAFKA_EVENT_LOG => self.handle_log_event(ev.ptr()),
-                rdsys::RD_KAFKA_EVENT_STATS => self.handle_stats_event(ev.ptr()),
+                rdsys::RD_KAFKA_EVENT_LOG => {
+                    self.handle_log_event(ev.ptr());
+                    return EventPollResult::EventConsumed;
+                }
+                rdsys::RD_KAFKA_EVENT_STATS => {
+                    self.handle_stats_event(ev.ptr());
+                    return EventPollResult::EventConsumed;
+                }
                 rdsys::RD_KAFKA_EVENT_ERROR => {
                     // rdkafka reports consumer errors via RD_KAFKA_EVENT_ERROR but producer errors gets
                     // embedded on the ack returned via RD_KAFKA_EVENT_DR. Hence we need to return this event
                     // for the consumer case in order to return the error to the user.
                     self.handle_error_event(ev.ptr());
-                    return Some(ev);
+                    return EventPollResult::Event(ev);
                 }
                 rdsys::RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH => {
                     if C::ENABLE_REFRESH_OAUTH_TOKEN {
                         self.handle_oauth_refresh_event(ev.ptr());
                     }
+                    return EventPollResult::EventConsumed;
                 }
                 _ => {
-                    return Some(ev);
+                    return EventPollResult::Event(ev);
                 }
             }
         }
-        None
+        EventPollResult::None
     }
 
     fn handle_log_event(&self, event: *mut RDKafkaEvent) {

src/config.rs

@@ -23,7 +23,7 @@
 //! [librdkafka-config]: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
 
 use std::collections::HashMap;
-use std::ffi::{CStr, CString};
+use std::ffi::CString;
 use std::iter::FromIterator;
 use std::os::raw::c_char;
 use std::ptr;
@@ -150,10 +150,9 @@ impl NativeClientConfig {
         }
 
         // Convert the C string to a Rust string.
-        Ok(CStr::from_bytes_with_nul(&buf)
-            .unwrap()
-            .to_string_lossy()
-            .into())
+        Ok(String::from_utf8_lossy(&buf)
+            .trim_matches(char::from(0))
+            .to_string())
     }
 }