Fix BaseProducer::flush behavior and tests (#777)

Fix BaseProducer::flush behavior and tests. The flush function now correctly calls poll until all messages have been processed.

Also:
* Make it easier to pass extra arguments to the test suite
  (`./test_suite.sh -- --nocapture` for example);
* Implement fmt::Debug for Metadata;
* Set up logging during producer tests.
* Fix docs warnings and format.
* Apply suggestions from code review
* Fix version number format in cicd config

Author: fede1024

.github/workflows/ci.yml

@@ -75,14 +75,10 @@ jobs:
       max-parallel: 1
       matrix:
         include:
-          - confluent-version: 7.9.1
-            kafka-version: 3.9
-          - 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
+          - kafka-version: "4.0"
+          - kafka-version: "3.9"
+          - kafka-version: "3.8"
+          - kafka-version: "3.7"
     runs-on: ubuntu-24.04
     steps:
       - uses: actions/checkout@v4
@@ -94,6 +90,5 @@ jobs:
       # - run: sudo apt-get install -qy valgrind  # Valgrind currently disabled in testing
       - run: ./test_suite.sh
         env:
-          CONFLUENT_VERSION: ${{ matrix.confluent-version }}
           KAFKA_VERSION: ${{ matrix.kafka-version }}
           TERM: xterm-256color

docker-compose.yaml

@@ -1,18 +1,23 @@
 services:
   kafka:
-    image: confluentinc/cp-kafka:${CONFLUENT_VERSION:-7.9.1}
+    image: bitnami/kafka:${KAFKA_VERSION:-4.0}
     environment:
-      CLUSTER_ID: "test-cluster"
-      KAFKA_NODE_ID: 0
-      KAFKA_PROCESS_ROLES: broker,controller
-      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
-      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
-      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@localhost:9093
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
-      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
-      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
-      KAFKA_NUM_PARTITIONS: 3
+      # Enable KRaft mode (combined broker and controller)
+      - KAFKA_CFG_NODE_ID=0
+      - KAFKA_CFG_BROKER_ID=0 # In KRaft, this should be the same as the node ID
+      - KAFKA_CFG_PROCESS_ROLES=broker,controller
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@localhost:9093
+      
+      # Bitnami defaults
+      - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=1
+      - KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
+      - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=1
+      - KAFKA_CFG_NUM_PARTITIONS=3
+
+      # This is a Bitnami-specific variable to disable ZooKeeper
+      - KAFKA_KRAFT_ENABLED=true
     ports: ["9092:9092"]

rdkafka-sys/build.rs

@@ -79,13 +79,13 @@ fn main() {
             println!("cargo:rustc-link-lib=static=rdkafka");
             println!("cargo:root={}", rdkafka_dir);
         } else {
-            eprintln!("Path to DEP_LIBRDKAFKA_STATIC_ROOT not set. Static linking failed. Exiting.");
+            eprintln!(
+                "Path to DEP_LIBRDKAFKA_STATIC_ROOT not set. Static linking failed. Exiting."
+            );
             process::exit(1);
         }
         eprintln!("librdkafka will be linked statically using prebuilt binaries");
-
-    }
-        else {
+    } else {
         // Ensure that we are in the right directory
         let rdkafkasys_root = Path::new("rdkafka-sys");
         if rdkafkasys_root.exists() {
@@ -226,9 +226,9 @@ fn build_librdkafka() {
         // Since we're not actually installing to /usr or /usr/local, there's no
         // harm to always using "lib" here.
         .define("CMAKE_INSTALL_LIBDIR", "lib")
-	// CMake 4.0.0 drops support for 3.2 compatibility, which is
-	// required by librdkafka 2.3.0.
-	.define("CMAKE_POLICY_VERSION_MINIMUM", "3.5");
+        // CMake 4.0.0 drops support for 3.2 compatibility, which is
+        // required by librdkafka 2.3.0.
+        .define("CMAKE_POLICY_VERSION_MINIMUM", "3.5");
 
     if env::var("CARGO_FEATURE_LIBZ").is_ok() {
         config.define("WITH_ZLIB", "1");

src/admin.rs

@@ -224,7 +224,7 @@ impl<C: ClientContext> AdminClient<C> {
     /// 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`]
+    /// in the specified partition will be deleted. Use offset [`crate::Offset::End`]
     /// to delete all records in the partition.
     ///
     /// Returns a topic partition list describing the result of the deletion. If

src/consumer/stream_consumer.rs

@@ -1,4 +1,4 @@
-//! High-level consumers with a [`Stream`](futures_util::Stream) interface.
+//! High-level consumers with a [`Stream`] interface.
 
 use std::marker::PhantomData;
 use std::os::raw::c_void;
@@ -172,16 +172,16 @@ impl<'a, C: ConsumerContext> Stream for MessageStream<'a, C> {
     }
 }
 
-impl<'a, C: ConsumerContext> Drop for MessageStream<'a, C> {
+impl<C: ConsumerContext> Drop for MessageStream<'_, C> {
     fn drop(&mut self) {
         self.wakers.unregister(self.slot);
     }
 }
 
-/// A high-level consumer with a [`Stream`](futures_util::Stream) interface.
+/// A high-level consumer with a [`Stream`] interface.
 ///
 /// This consumer doesn't need to be polled explicitly. Extracting an item from
-/// the stream returned by the [`stream`](StreamConsumer::stream) will
+/// the stream returned by [`stream`](StreamConsumer::stream) will
 /// implicitly poll the underlying Kafka consumer.
 ///
 /// If you activate the consumer group protocol by calling

src/message.rs

@@ -320,7 +320,7 @@ unsafe impl KafkaDrop for RDKafkaMessage {
     const DROP: unsafe extern "C" fn(*mut Self) = no_op;
 }
 
-impl<'a> fmt::Debug for BorrowedMessage<'a> {
+impl fmt::Debug for BorrowedMessage<'_> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
@@ -421,7 +421,7 @@ impl<'a> BorrowedMessage<'a> {
     }
 }
 
-impl<'a> Message for BorrowedMessage<'a> {
+impl Message for BorrowedMessage<'_> {
     type Headers = BorrowedHeaders;
 
     fn key(&self) -> Option<&[u8]> {
@@ -488,8 +488,8 @@ impl<'a> Message for BorrowedMessage<'a> {
     }
 }
 
-unsafe impl<'a> Send for BorrowedMessage<'a> {}
-unsafe impl<'a> Sync for BorrowedMessage<'a> {}
+unsafe impl Send for BorrowedMessage<'_> {}
+unsafe impl Sync for BorrowedMessage<'_> {}
 
 //
 // ********** OWNED MESSAGE **********

src/metadata.rs

@@ -1,6 +1,7 @@
 //! Cluster metadata.
 
 use std::ffi::CStr;
+use std::fmt;
 use std::slice;
 
 use rdkafka_sys as rdsys;
@@ -33,6 +34,16 @@ impl MetadataBroker {
     }
 }
 
+impl fmt::Debug for MetadataBroker {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("MetadataBroker")
+            .field("id", &self.id())
+            .field("host", &self.host())
+            .field("port", &self.port())
+            .finish()
+    }
+}
+
 /// Partition metadata information.
 pub struct MetadataPartition(RDKafkaMetadataPartition);
 
@@ -69,6 +80,21 @@ impl MetadataPartition {
     }
 }
 
+impl fmt::Debug for MetadataPartition {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut debug_struct = f.debug_struct("MetadataPartition");
+        debug_struct.field("id", &self.id());
+        if let Some(err) = self.error() {
+            debug_struct.field("error", &err);
+        }
+        debug_struct
+            .field("leader", &self.leader())
+            .field("replicas", &self.replicas())
+            .field("isr", &self.isr()) // In-Sync Replicas
+            .finish()
+    }
+}
+
 /// Topic metadata information.
 pub struct MetadataTopic(RDKafkaMetadataTopic);
 
@@ -103,6 +129,18 @@ impl MetadataTopic {
     }
 }
 
+impl fmt::Debug for MetadataTopic {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut debug_struct = f.debug_struct("MetadataTopic");
+        debug_struct.field("name", &self.name());
+        if let Some(err) = self.error() {
+            debug_struct.field("error", &err);
+        }
+        debug_struct.field("partitions", &self.partitions());
+        debug_struct.finish()
+    }
+}
+
 /// Metadata container.
 ///
 /// This structure wraps the metadata pointer returned by rdkafka-sys, and
@@ -159,5 +197,16 @@ impl Metadata {
     }
 }
 
+impl fmt::Debug for Metadata {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Metadata")
+            .field("orig_broker_name", &self.orig_broker_name())
+            .field("orig_broker_id", &self.orig_broker_id())
+            .field("brokers", &self.brokers())
+            .field("topics", &self.topics())
+            .finish()
+    }
+}
+
 unsafe impl Send for Metadata {}
 unsafe impl Sync for Metadata {}

src/mocking.rs

@@ -383,7 +383,7 @@ where
     }
 }
 
-impl<'c, C> Drop for MockCluster<'c, C>
+impl<C> Drop for MockCluster<'_, C>
 where
     C: ClientContext,
 {

src/producer/base_producer.rs

@@ -507,23 +507,24 @@ where
 
     // As this library uses the rdkafka Event API, flush will not call rd_kafka_poll() but instead wait for
     // the librdkafka-handled message count to reach zero. Runs until value reaches zero or timeout.
+    // https://github.com/confluentinc/librdkafka/blob/c024ac13daf98667de2b8724986e97f489644c15/src/rdkafka.c#L4542-L4551
     fn flush<T: Into<Timeout>>(&self, timeout: T) -> KafkaResult<()> {
         let deadline: Deadline = timeout.into().into();
-        while self.in_flight_count() > 0 && !deadline.elapsed() {
-            let ret: RDKafkaRespErr = unsafe {
-                rdsys::rd_kafka_flush(self.native_ptr(), deadline.remaining_millis_i32())
-            };
-            if let Deadline::Never = &deadline {
-                self.poll(Timeout::After(Duration::ZERO));
-            } else {
-                self.poll(&deadline);
-            }
-
-            if ret.is_error() {
-                return Err(KafkaError::Flush(ret.into()));
+        loop {
+            match unsafe { rdsys::rd_kafka_flush(self.native_ptr(), 0) } {
+                rdsys::rd_kafka_resp_err_t::RD_KAFKA_RESP_ERR_NO_ERROR => {
+                    // Flush completed
+                    return Ok(());
+                }
+                to @ rdsys::rd_kafka_resp_err_t::RD_KAFKA_RESP_ERR__TIMED_OUT => {
+                    if deadline.elapsed() {
+                        return Err(KafkaError::Flush(to.into()));
+                    }
+                    self.poll(deadline.remaining().min(Duration::from_millis(100)));
+                }
+                e => return Err(KafkaError::Flush(e.into())),
             };
         }
-        Ok(())
     }
 
     fn purge(&self, flags: PurgeConfig) {

src/producer/mod.rs

@@ -205,7 +205,7 @@ pub trait ProducerContext<Part: Partitioner = NoCustomPartitioner>: ClientContex
     /// If custom partitioner is not used then `partitioner` configuration property is used (or its default).
     ///
     /// sticky.partitioning.linger.ms must be 0 to run custom partitioner for messages with null key.
-    /// See https://github.com/confluentinc/librdkafka/blob/081fd972fa97f88a1e6d9a69fc893865ffbb561a/src/rdkafka_msg.c#L1192-L1196
+    /// See <https://github.com/confluentinc/librdkafka/blob/081fd972fa97f88a1e6d9a69fc893865ffbb561a/src/rdkafka_msg.c#L1192-L1196>
     fn get_custom_partitioner(&self) -> Option<&Part> {
         None
     }