From a3bfb6ddb217102ace47bb3c1a0db29d06720712 Mon Sep 17 00:00:00 2001 From: florin akermann Date: Tue, 29 Apr 2025 22:38:50 +0200 Subject: [PATCH] Refactor PartitionEOF to carry topic, partition, and offset Previously `KafkaError::PartitionEOF` carried only the partition as an `i32`. For a consumer subscribed to more than one topic, that partition number is ambiguous. From the EOF signal alone there is no way to tell which topic-partition reached the end. This change replaces the payload with a `TopicPartitionOffset` struct (topic, partition, offset). Callers now get the full coordinates on every EOF signal. This makes it possible to track, from the EOF stream alone, whether the consumer reached the end for all of its topic-partitions. At EOF the offset is the log-end position, so the catch-up point comes for free. A concrete use case is an app that builds and serves state derived from Kafka topics. It can now detect when it has fully caught up: EOF observed on every relevant topic-partition. Only then does it mark itself ready. This enables smoother blue/green deployments. A freshly started instance warms up its state first. It begins serving traffic only once it is up to speed. This catch-up detection is also achievable today without this change. You can compare the consumed position against the high watermark per partition (e.g. via `fetch_watermarks`). The value here is ergonomics. The EOF approach is event-driven. It needs no separate watermark polling loop. It avoids the race between reading the watermark and consuming up to it. EOF is a point-in-time signal. It suits an initial warm-up gate rather than a continuous liveness check. --- changelog.md | 7 +++- src/consumer/base_consumer.rs | 14 ++++++- src/error.rs | 4 +- src/message.rs | 13 +++++- src/util.rs | 21 ++++++++++ tests/base_consumer.rs | 76 ++++++++++++++++++++++++++++++++++- 6 files changed, 127 insertions(+), 8 deletions(-) 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), + } +}