fix: start timestamp

edgeandnode · Oct 17, 2024 · e988fb8 · e988fb8
1 parent 117ebca
commit e988fb8
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/src/bin/main.rs b/src/bin/main.rs
@@ -66,7 +66,8 @@ async fn run() -> anyhow::Result<()> {
     )?
     .create()?;
 
-    let start_timestamp = latest_sink_timestamp(&consumer).await?.unwrap_or(0);
+    let start_timestamp =
+        latest_sink_timestamp(&consumer).await?.unwrap_or(0) + AGGREGATION_INTERVAL_MILLIS;
     consumer.unassign().context("unassign sinks")?;
     tracing::info!(start_timestamp = print_unix_millis(start_timestamp)?);