Merge pull request #246 from yra1029/Add_producer_timestamp

feat: Added the support for the new format of producer message with timestamp support

Author: johnward

Cargo.toml

@@ -22,6 +22,7 @@ flate2 = { version = "1.0", optional = true }
 openssl = { version = "0.10", optional = true }
 openssl-sys = { version = "0.9", optional = true }
 snap = { version = "1.1", optional = true }
+chrono = {version = "0.4", optional = true}
 thiserror = "1.0"
 tracing = "0.1"
 
@@ -38,5 +39,6 @@ default = ["snappy", "gzip", "security"]
 snappy = ["snap"]
 gzip = ["flate2"]
 security = ["openssl", "openssl-sys"]
+producer_timestamp = ["chrono"]
 nightly = []
 integration_tests = []

src/client/mod.rs

@@ -19,6 +19,12 @@ use crate::protocol::list_offset::ListOffsetVersion;
 pub use crate::utils::PartitionOffset;
 use crate::utils::TimestampedPartitionOffset;
 
+#[cfg(feature = "producer_timestamp")]
+pub use crate::protocol::produce::ProducerTimestamp;
+
+#[cfg(not(feature = "producer_timestamp"))]
+use crate::protocol::produce::ProducerTimestamp;
+
 #[cfg(feature = "security")]
 pub use self::network::SecurityConfig;
 
@@ -78,6 +84,9 @@ pub const DEFAULT_RETRY_MAX_ATTEMPTS: u32 = 120_000 / DEFAULT_RETRY_BACKOFF_TIME
 /// The default value for `KafkaClient::set_connection_idle_timeout(..)`
 pub const DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS: u64 = 540_000;
 
+/// The default value for `KafkaClient::set_producer_timestamp(..)`
+pub(crate) const DEFAULT_PRODUCER_TIMESTAMP: Option<ProducerTimestamp> = None;
+
 /// Client struct keeping track of brokers and topic metadata.
 ///
 /// Implements methods described by the [Kafka Protocol](http://kafka.apache.org/protocol.html).
@@ -122,6 +131,9 @@ struct ClientConfig {
     // ~ the number of repeated retry attempts; prevents endless
     // repetition of a retry attempt
     retry_max_attempts: u32,
+    // ~ producer's message timestamp option CreateTime/LogAppendTime
+    #[allow(unused)]
+    producer_timestamp: Option<ProducerTimestamp>,
 }
 
 // --------------------------------------------------------------------
@@ -408,6 +420,7 @@ impl KafkaClient {
                 offset_storage: DEFAULT_GROUP_OFFSET_STORAGE,
                 retry_backoff_time: Duration::from_millis(DEFAULT_RETRY_BACKOFF_TIME_MILLIS),
                 retry_max_attempts: DEFAULT_RETRY_MAX_ATTEMPTS,
+                producer_timestamp: DEFAULT_PRODUCER_TIMESTAMP,
             },
             conn_pool: network::Connections::new(
                 default_conn_rw_timeout(),
@@ -477,6 +490,7 @@ impl KafkaClient {
                 offset_storage: DEFAULT_GROUP_OFFSET_STORAGE,
                 retry_backoff_time: Duration::from_millis(DEFAULT_RETRY_BACKOFF_TIME_MILLIS),
                 retry_max_attempts: DEFAULT_RETRY_MAX_ATTEMPTS,
+                producer_timestamp: DEFAULT_PRODUCER_TIMESTAMP,
             },
             conn_pool: network::Connections::new_with_security(
                 default_conn_rw_timeout(),
@@ -722,6 +736,31 @@ impl KafkaClient {
         self.conn_pool.idle_timeout()
     }
 
+    #[cfg(feature = "producer_timestamp")]
+    /// Sets the compression algorithm to use when sending out messages.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// use kafka::client::{Compression, KafkaClient};
+    ///
+    /// let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
+    /// client.load_metadata_all().unwrap();
+    /// client.set_producer_timestamp(Timestamp::CreateTime);
+    /// ```
+    #[inline]
+    pub fn set_producer_timestamp(&mut self, producer_timestamp: Option<ProducerTimestamp>) {
+        self.config.producer_timestamp = producer_timestamp;
+    }
+
+    #[cfg(feature = "producer_timestamp")]
+    /// Retrieves the current `KafkaClient::producer_timestamp` setting.
+    #[inline]
+    #[must_use]
+    pub fn producer_timestamp(&self) -> Option<ProducerTimestamp> {
+        self.config.producer_timestamp
+    }
+
     /// Provides a view onto the currently loaded metadata of known .
     ///
     /// # Examples
@@ -1455,6 +1494,8 @@ impl KafkaClientInternals for KafkaClient {
                             correlation,
                             &config.client_id,
                             config.compression,
+                            #[cfg(feature = "producer_timestamp")]
+                            config.producer_timestamp,
                         )
                     })
                     .add(msg.topic, msg.partition, msg.key, msg.value),

src/producer.rs

@@ -68,6 +68,9 @@ use std::slice::from_ref;
 use std::time::Duration;
 use twox_hash::XxHash32;
 
+#[cfg(feature = "producer_timestamp")]
+use crate::protocol::produce::ProducerTimestamp;
+
 #[cfg(feature = "security")]
 use crate::client::SecurityConfig;
 
@@ -360,6 +363,8 @@ pub struct Builder<P = DefaultPartitioner> {
     partitioner: P,
     security_config: Option<SecurityConfig>,
     client_id: Option<String>,
+    #[cfg(feature = "producer_timestamp")]
+    producer_timestamp: Option<ProducerTimestamp>,
 }
 
 impl Builder {
@@ -376,6 +381,8 @@ impl Builder {
             partitioner: DefaultPartitioner::default(),
             security_config: None,
             client_id: None,
+            #[cfg(feature = "producer_timestamp")]
+            producer_timestamp: None,
         };
         if let Some(ref c) = b.client {
             b.compression = c.compression();
@@ -437,6 +444,16 @@ impl Builder {
         self.client_id = Some(client_id);
         self
     }
+
+    #[cfg(feature = "producer_timestamp")]
+    /// Sets the compression algorithm to use when sending out data.
+    ///
+    /// See `KafkaClient::set_producer_timestamp`.
+    #[must_use]
+    pub fn with_timestamp(mut self, timestamp: ProducerTimestamp) -> Self {
+        self.producer_timestamp = Some(timestamp);
+        self
+    }
 }
 
 impl<P> Builder<P> {
@@ -453,6 +470,8 @@ impl<P> Builder<P> {
             partitioner,
             security_config: None,
             client_id: None,
+            #[cfg(feature = "producer_timestamp")]
+            producer_timestamp: None,
         }
     }
 
@@ -484,6 +503,8 @@ impl<P> Builder<P> {
         // ~ apply configuration settings
         client.set_compression(self.compression);
         client.set_connection_idle_timeout(self.conn_idle_timeout);
+        #[cfg(feature = "producer_timestamp")]
+        client.set_producer_timestamp(self.producer_timestamp);
         if let Some(client_id) = self.client_id {
             client.set_client_id(client_id);
         }