PartitionEOF: carry topic + offset so consumers can detect full catch-up

changelog.md

@@ -4,7 +4,12 @@ See also the [rdkafka-sys changelog](rdkafka-sys/changelog.md).
 
 ## Unreleased
 
-None
+* **Breaking change:** `KafkaError::PartitionEOF` now carries a
+  `TopicPartitionOffset` (topic, partition, and offset) instead of only the
+  partition `i32`, providing full context when a partition reaches EOF
+  ([#784]).
+
+[#784]: https://github.com/fede1024/rust-rdkafka/pull/784
 
 ## 0.39.0 (2026-01-25)
 

src/consumer/base_consumer.rs

@@ -26,7 +26,7 @@ use crate::log::trace;
 use crate::message::{BorrowedMessage, Message};
 use crate::metadata::Metadata;
 use crate::topic_partition_list::{Offset, TopicPartitionList};
-use crate::util::{cstr_to_owned, NativePtr, Timeout};
+use crate::util::{cstr_to_owned, NativePtr, Timeout, TopicPartitionOffset};
 
 /// A low-level consumer that requires manual polling.
 ///
@@ -259,8 +259,18 @@ where
             if rdkafka_err == rdsys::rd_kafka_resp_err_t::RD_KAFKA_RESP_ERR__PARTITION_EOF {
                 let tp_ptr = unsafe { rdsys::rd_kafka_event_topic_partition(event.ptr()) };
                 let partition = unsafe { (*tp_ptr).partition };
+                let topic = unsafe {
+                    CStr::from_ptr((*tp_ptr).topic)
+                        .to_string_lossy()
+                        .into_owned()
+                };
+                let offset = unsafe { (*tp_ptr).offset };
                 unsafe { rdsys::rd_kafka_topic_partition_destroy(tp_ptr) };
-                Some(KafkaError::PartitionEOF(partition))
+                Some(KafkaError::PartitionEOF(TopicPartitionOffset {
+                    topic,
+                    partition,
+                    offset,
+                }))
             } else if unsafe { rdsys::rd_kafka_event_error_is_fatal(event.ptr()) } != 0 {
                 Some(KafkaError::MessageConsumptionFatal(rdkafka_err.into()))
             } else {

src/error.rs

@@ -9,7 +9,7 @@ use std::sync::Arc;
 use rdkafka_sys as rdsys;
 use rdkafka_sys::types::*;
 
-use crate::util::{KafkaDrop, NativePtr};
+use crate::util::{KafkaDrop, NativePtr, TopicPartitionOffset};
 
 // Re-export rdkafka error code
 pub use rdsys::types::RDKafkaErrorCode;
@@ -170,7 +170,7 @@ pub enum KafkaError {
     /// Offset fetch failed.
     OffsetFetch(RDKafkaErrorCode),
     /// End of partition reached.
-    PartitionEOF(i32),
+    PartitionEOF(TopicPartitionOffset),
     /// Pause/Resume failed.
     PauseResume(String),
     /// Rebalance failed.

src/message.rs

@@ -14,7 +14,7 @@ use rdkafka_sys::types::*;
 
 use crate::admin::NativeEvent;
 use crate::error::{IsError, KafkaError, KafkaResult};
-use crate::util::{self, millis_to_epoch, KafkaDrop, NativePtr};
+use crate::util::{self, millis_to_epoch, KafkaDrop, NativePtr, TopicPartitionOffset};
 
 /// Timestamp of a Kafka message.
 #[derive(Debug, PartialOrd, Ord, PartialEq, Eq, Clone, Copy)]
@@ -346,7 +346,16 @@ impl<'a> BorrowedMessage<'a> {
         if ptr.err.is_error() {
             let err = match ptr.err {
                 rdsys::rd_kafka_resp_err_t::RD_KAFKA_RESP_ERR__PARTITION_EOF => {
-                    KafkaError::PartitionEOF(ptr.partition)
+                    let topic = unsafe {
+                        CStr::from_ptr(rdsys::rd_kafka_topic_name(ptr.rkt))
+                            .to_string_lossy()
+                            .into_owned()
+                    };
+                    KafkaError::PartitionEOF(TopicPartitionOffset {
+                        topic,
+                        partition: ptr.partition,
+                        offset: ptr.offset,
+                    })
                 }
                 e => KafkaError::MessageConsumption(e.into()),
             };

src/util.rs

@@ -37,6 +37,27 @@ pub(crate) enum Deadline {
     Never,
 }
 
+/// Represents the coordinates of a kafka record
+#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
+pub struct TopicPartitionOffset {
+    /// The name of the Kafka topic.
+    pub topic: String,
+    /// The partition within the Kafka topic.
+    pub partition: i32,
+    /// The offset within the specified Kafka partition.
+    pub offset: i64,
+}
+
+impl fmt::Display for TopicPartitionOffset {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "Topic: {}, Partition: {}, Offset: {}",
+            self.topic, self.partition, self.offset
+        )
+    }
+}
+
 impl Deadline {
     // librdkafka's flush api requires an i32 millisecond timeout
     const MAX_FLUSH_DURATION: Duration = Duration::from_millis(i32::MAX as u64);

tests/base_consumer.rs

@@ -10,7 +10,8 @@ use std::time::{Duration, Instant};
 use rdkafka::admin::AdminOptions;
 use rdkafka::client::ClientContext;
 use rdkafka::consumer::{BaseConsumer, Consumer, ConsumerContext, Rebalance};
-use rdkafka::error::{KafkaError, RDKafkaErrorCode};
+use rdkafka::error::{KafkaError, KafkaResult, RDKafkaErrorCode};
+use rdkafka::message::BorrowedMessage;
 use rdkafka::topic_partition_list::{Offset, TopicPartitionList};
 use rdkafka::util::{current_time_millis, Timeout};
 use rdkafka::{ClientConfig, Message, Timestamp};
@@ -887,3 +888,76 @@ async fn test_consumer_rebalance_callbacks() {
     );
     assert_eq!(assign2.partitions[0].0, topic_name);
 }
+
+fn poll_skipping_transient<C: ConsumerContext>(
+    consumer: &BaseConsumer<C>,
+    timeout: Duration,
+) -> KafkaResult<BorrowedMessage<'_>> {
+    loop {
+        match consumer.poll(Timeout::from(timeout)) {
+            Some(Err(KafkaError::MessageConsumption(
+                RDKafkaErrorCode::BrokerTransportFailure | RDKafkaErrorCode::AllBrokersDown,
+            ))) => continue,
+            Some(result) => return result,
+            None => panic!("No message or error received within timeout"),
+        }
+    }
+}
+
+#[tokio::test]
+async fn test_partition_eof_error_details() {
+    init_test_logger();
+    let kafka_context = KafkaContext::shared()
+        .await
+        .expect("could not create kafka context");
+    let topic_name = rand_test_topic("test_partition_eof_error_details");
+    let message_count = 5usize;
+
+    let admin_client = admin::create_admin_client(&kafka_context.bootstrap_servers)
+        .await
+        .expect("could not create admin client");
+    admin_client
+        .create_topics(
+            &admin::new_topic_vec(&topic_name, Some(1)),
+            &AdminOptions::default(),
+        )
+        .await
+        .expect("could not create topic");
+
+    let producer = producer::future_producer::create_producer(&kafka_context.bootstrap_servers)
+        .await
+        .expect("could not create producer");
+    produce_messages_to_partition(&producer, &topic_name, message_count, 0).await;
+
+    let consumer = utils::consumer::create_base_consumer(
+        &kafka_context.bootstrap_servers,
+        &rand_test_group(),
+        Some(&[("enable.partition.eof", "true")]),
+    )
+    .expect("could not create base consumer");
+
+    let mut tpl = TopicPartitionList::new();
+    tpl.add_partition_offset(&topic_name, 0, Offset::Beginning)
+        .unwrap();
+    consumer.assign(&tpl).unwrap();
+
+    let timeout = Duration::from_secs(5);
+
+    for received in 0..message_count {
+        let message = poll_skipping_transient(&consumer, timeout)
+            .unwrap_or_else(|e| panic!("Error receiving message: {:?}", e));
+        assert_eq!(message.offset(), received as i64);
+        assert_eq!(message.partition(), 0);
+        assert_eq!(message.topic(), topic_name);
+    }
+
+    match poll_skipping_transient(&consumer, timeout) {
+        Err(KafkaError::PartitionEOF(tpo)) => {
+            assert_eq!(tpo.topic, topic_name);
+            assert_eq!(tpo.partition, 0);
+            assert_eq!(tpo.offset, message_count as i64);
+        }
+        Ok(_) => panic!("Expected PartitionEOF error, got message"),
+        Err(e) => panic!("Expected PartitionEOF error, got: {:?}", e),
+    }
+}