Fix integration tests

Fixed several issues with the current test suite:

* Missing cmake dependency
* Rust version being too old for some dependencies
* Missing null pointer check in src/groups.rs
* Test timeout being too strict
* CI test jobs not working in parallel

This also fixes a few compiler warnings (unused imports, unused code,
etc.) and formatting issues.

.github/workflows/ci.yml

@@ -7,14 +7,14 @@ 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
@@ -38,7 +38,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,43 +51,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"]

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/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")]

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,7 +150,9 @@ impl NativeClientConfig {
         }
 
         // Convert the C string to a Rust string.
-        Ok(String::from_utf8_lossy(&buf).trim_matches(char::from(0)).to_string())
+        Ok(String::from_utf8_lossy(&buf)
+            .trim_matches(char::from(0))
+            .to_string())
     }
 }
 

src/groups.rs

@@ -84,6 +84,9 @@ impl GroupInfo {
 
     /// Returns the members of the group.
     pub fn members(&self) -> &[GroupMemberInfo] {
+        if self.0.members.is_null() {
+            return &[];
+        }
         unsafe {
             slice::from_raw_parts(
                 self.0.members as *const GroupMemberInfo,

src/lib.rs

@@ -176,7 +176,7 @@
 //!
 //! ### 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.
 //!

src/mocking.rs

@@ -39,7 +39,9 @@ use crate::ClientContext;
 ///   In this case, we **must neither** destroy the mock cluster in `MockCluster`'s `drop()`,
 ///   **nor** outlive the `Client` from which the reference is obtained, hence the lifetime.
 enum MockClusterClient<'c, C: ClientContext> {
+    #[allow(dead_code)]
     Owned(Client<C>),
+    #[allow(dead_code)]
     Borrowed(&'c Client<C>),
 }
 

src/util.rs

@@ -251,24 +251,13 @@ impl fmt::Display for ErrBuf {
     }
 }
 
-pub(crate) trait WrappedCPointer {
-    type Target;
-
-    fn ptr(&self) -> *mut Self::Target;
-
-    fn is_null(&self) -> bool {
-        self.ptr().is_null()
-    }
+pub(crate) trait AsCArray<T> {
+    fn as_c_array(&self) -> *mut *mut T;
 }
 
-/// Converts a container into a C array.
-pub(crate) trait AsCArray<T: WrappedCPointer> {
-    fn as_c_array(&self) -> *mut *mut T::Target;
-}
-
-impl<T: WrappedCPointer> AsCArray<T> for Vec<T> {
-    fn as_c_array(&self) -> *mut *mut T::Target {
-        self.as_ptr() as *mut *mut T::Target
+impl<T: KafkaDrop> AsCArray<T> for Vec<NativePtr<T>> {
+    fn as_c_array(&self) -> *mut *mut T {
+        self.as_ptr() as *mut *mut T
     }
 }
 
@@ -297,17 +286,6 @@ pub(crate) unsafe trait KafkaDrop {
     const DROP: unsafe extern "C" fn(*mut Self);
 }
 
-impl<T> WrappedCPointer for NativePtr<T>
-where
-    T: KafkaDrop,
-{
-    type Target = T;
-
-    fn ptr(&self) -> *mut T {
-        self.ptr.as_ptr()
-    }
-}
-
 impl<T> Deref for NativePtr<T>
 where
     T: KafkaDrop,
@@ -340,6 +318,7 @@ where
     }
 }
 
+#[allow(dead_code)]
 pub(crate) struct OnDrop<F>(pub F)
 where
     F: Fn();

test_suite.sh

@@ -29,8 +29,12 @@ run_with_valgrind() {
 # Initialize.
 
 git submodule update --init
-docker-compose up -d
-cargo test
+docker compose up -d --wait
+
+# Run integration tests
+
+RUST_LOG=1 RUST_BACKTRACE=1 cargo test
+
 
 # Run unit tests.
 

tests/test_admin.rs

@@ -119,7 +119,7 @@ fn verify_delete(topic: &str) {
 #[tokio::test]
 async fn test_topics() {
     let admin_client = create_admin_client();
-    let opts = AdminOptions::new().operation_timeout(Some(Duration::from_secs(1)));
+    let opts = AdminOptions::new().operation_timeout(Some(Duration::from_secs(30)));
 
     // Verify that topics are created as specified, and that they can later
     // be deleted.