diff --git a/changelog.md b/changelog.md index 03b69f046..000cf7a21 100644 --- a/changelog.md +++ b/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) diff --git a/src/consumer/base_consumer.rs b/src/consumer/base_consumer.rs index f69ccffc0..88310439f 100644 --- a/src/consumer/base_consumer.rs +++ b/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 { diff --git a/src/error.rs b/src/error.rs index 634fe9eee..364ea5886 100644 --- a/src/error.rs +++ b/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. diff --git a/src/message.rs b/src/message.rs index e72741219..32aec652f 100644 --- a/src/message.rs +++ b/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()), }; diff --git a/src/util.rs b/src/util.rs index 719033780..a00cbce46 100644 --- a/src/util.rs +++ b/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); diff --git a/tests/base_consumer.rs b/tests/base_consumer.rs index 484675d29..63f120740 100644 --- a/tests/base_consumer.rs +++ b/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( + consumer: &BaseConsumer, + timeout: Duration, +) -> KafkaResult> { + 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), + } +}