Create a new Delivery struct that includes Timestamp (fede1024#555)

* Add `Timestamp` to the `OwnedDeliveryResult` tuple.

**Note:** This is an API breaking change.

Currently the `Timestamp` value is being dropped as part of the
conversion from `BorrowedMessage` to `OwnedDeliveryResult`, however this
value is updated once the message has been published and callers should
have access to the updated value. As far as I can tell, this is the
**only** other value that is updated as part of publishing.

An alternative to expanding the tuple would be to replace it with a
struct. This is a bit more disruptive now, but would making adding
values in the future a backwards compatible change. So it depends how
likely it is that there will be further new values.

This could also be done in a way that isn't a breaking change, but would
involve duplicating functions and copying some types, i.e.
`DeliveryFuture`. It doesn't seem like that level of complexity is
justified given the pre 1.0 status.

* Switch from tuple to struct in OwnedDeliveryResult.

* Remove unused import.

* Rename struct.

* Add breaking change notice to the changelog.

* Respect existing line lengths.

Author: gorzell

changelog.md

@@ -10,6 +10,10 @@ See also the [rdkafka-sys changelog](rdkafka-sys/changelog.md).
 * Address wakeup races introduced by pivoting to the event API.
 * Update `BaseProducer::poll` to not return early, and instead continue
   looping until the passed timeout is reached.
+* **Breaking change.** Change signature for `OwnedDeliveryResult`. The 
+`Ok` variant is now a `Delivery` struct, rather than a tuple. This allows
+or including `Timestamp` as a result field. It means that adding values 
+in the future will not require a breaking change.
 
 ## 0.36.2 (2024-01-16)
 

src/message.rs

@@ -17,7 +17,7 @@ use crate::error::{IsError, KafkaError, KafkaResult};
 use crate::util::{self, millis_to_epoch, KafkaDrop, NativePtr};
 
 /// Timestamp of a Kafka message.
-#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+#[derive(Debug, PartialOrd, Ord, PartialEq, Eq, Clone, Copy)]
 pub enum Timestamp {
     /// Timestamp not available.
     NotAvailable,

src/producer/future_producer.rs

@@ -132,14 +132,23 @@ pub struct FutureProducerContext<C: ClientContext + 'static> {
     wrapped_context: C,
 }
 
-/// Represents the result of message production as performed from the
-/// `FutureProducer`.
+/// Contains information about a successfully delivered message
+#[derive(Debug, PartialEq, Eq)]
+pub struct Delivery {
+    /// The partition the message was delivered to
+    pub partition: i32,
+    /// The offset within the partition
+    pub offset: i64,
+    /// The timestamp associated with the message
+    pub timestamp: Timestamp,
+}
+
+/// Represents the result of message production as performed from the FutureProducer.
 ///
-/// If message delivery was successful, `OwnedDeliveryResult` will return the
-/// partition and offset of the message. If the message failed to be delivered
-/// an error will be returned, together with an owned copy of the original
-/// message.
-pub type OwnedDeliveryResult = Result<(i32, i64), (KafkaError, OwnedMessage)>;
+/// If message delivery was successful, returns `DeliveredMessage` containing the partition,
+/// offset and timestamp. If the message failed to be delivered, returns the error and
+/// an owned copy of the original message.
+pub type OwnedDeliveryResult = Result<Delivery, (KafkaError, OwnedMessage)>;
 
 // Delegates all the methods calls to the wrapped context.
 impl<C: ClientContext + 'static> ClientContext for FutureProducerContext<C> {
@@ -183,7 +192,11 @@ where
         tx: Box<oneshot::Sender<OwnedDeliveryResult>>,
     ) {
         let owned_delivery_result = match *delivery_result {
-            Ok(ref message) => Ok((message.partition(), message.offset())),
+            Ok(ref message) => Ok(Delivery {
+                partition: message.partition(),
+                offset: message.offset(),
+                timestamp: message.timestamp(),
+            }),
             Err((ref error, ref message)) => Err((error.clone(), message.detach())),
         };
         let _ = tx.send(owned_delivery_result); // TODO: handle error

tests/test_high_producers.rs

@@ -11,6 +11,7 @@ use rdkafka::error::{KafkaError, RDKafkaErrorCode};
 use rdkafka::message::{Header, Headers, Message, OwnedHeaders};
 use rdkafka::producer::{FutureProducer, FutureRecord, Producer};
 use rdkafka::util::Timeout;
+use rdkafka::Timestamp;
 
 use crate::utils::*;
 
@@ -44,9 +45,10 @@ async fn test_future_producer_send() {
     let results: Vec<_> = results.collect().await;
     assert!(results.len() == 10);
     for (i, result) in results.into_iter().enumerate() {
-        let (partition, offset) = result.unwrap();
-        assert_eq!(partition, 1);
-        assert_eq!(offset, i as i64);
+        let delivered = result.unwrap();
+        assert_eq!(delivered.partition, 1);
+        assert_eq!(delivered.offset, i as i64);
+        assert!(delivered.timestamp < Timestamp::now());
     }
 }
 

tests/utils.rs

@@ -146,7 +146,7 @@ where
     let mut message_map = HashMap::new();
     for (id, future) in futures {
         match future.await {
-            Ok((partition, offset)) => message_map.insert((partition, offset), id),
+            Ok(delivered) => message_map.insert((delivered.partition, delivered.offset), id),
             Err((kafka_error, _message)) => panic!("Delivery failed: {}", kafka_error),
         };
     }