test: use pre-defined time for records when testing timestamp based o…

…ffset

tests/consumer.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use assert_matches::assert_matches;
-use chrono::{DateTime, Timelike, Utc};
+use chrono::{DateTime, TimeZone, Timelike, Utc};
 use futures::{Stream, StreamExt};
 use tokio::time::timeout;
 
@@ -435,35 +435,27 @@ async fn test_stream_consumer_start_timestamp_based_offset() {
         .await
         .unwrap();
 
-    fn now() -> DateTime<Utc> {
-        let now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .expect("system time before Unix epoch");
-        DateTime::from_timestamp(now.as_secs() as i64, now.subsec_nanos()).unwrap()
-    }
-
     let partition_client = Arc::new(
         client
             .partition_client(&topic, 0, UnknownTopicHandling::Retry)
             .await
             .unwrap(),
     );
-    let record_1 = record_with_timestamp_milliseconds(b"x", now());
+    let ts = Utc.timestamp_millis_opt(1337).unwrap();
+    let record_1 = record_with_timestamp_milliseconds(b"x", ts);
+    let record_2 = record_with_timestamp_milliseconds(b"y", ts + Duration::from_millis(100));
     partition_client
         .produce(vec![record_1.clone()], Compression::NoCompression)
         .await
         .unwrap();
-    tokio::time::sleep(Duration::from_millis(100)).await;
-    let ts = now();
 
-    let record_2 = record_with_timestamp_milliseconds(b"y", now());
     partition_client
         .produce(vec![record_2.clone()], Compression::NoCompression)
         .await
         .unwrap();
 
     let offset = partition_client
-        .get_offset(OffsetAt::Timestamp(ts))
+        .get_offset(OffsetAt::Timestamp(ts + Duration::from_millis(100)))
         .await
         .unwrap();
     assert_eq!(offset, 1);
@@ -507,8 +499,6 @@ fn record_with_timestamp_milliseconds(key: &[u8], timestamp: DateTime<Utc>) -> R
         key: Some(key.to_vec()),
         value: Some(b"hello kafka".to_vec()),
         headers: std::collections::BTreeMap::from([("foo".to_owned(), b"bar".to_vec())]),
-        timestamp: timestamp
-            .with_nanosecond(timestamp.nanosecond() / 1_000_000 * 1_000_000)
-            .unwrap(),
+        timestamp,
     }
 }