clean

Signed-off-by: Sidhant Kohli <[email protected]>

rust/numaflow-core/src/config.rs

@@ -1,14 +1,9 @@
-use futures::future::Lazy;
-use log::info;
-use monovertex::MonovertexConfig;
-use parking_lot::lock_api::MutexGuard;
-use parking_lot::RawMutex;
-use prometheus_client::registry::Registry;
 use std::env;
-use std::sync::{Mutex, OnceLock};
+use std::sync::OnceLock;
+
+use monovertex::MonovertexConfig;
 
 use crate::config::pipeline::PipelineConfig;
-use crate::metrics::GlobalMetrics;
 use crate::Error;
 use crate::Result;
 

rust/numaflow-core/src/config/monovertex.rs

@@ -1,18 +1,18 @@
 use std::time::Duration;
 
-use base64::Engine;
 use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
 use serde_json::from_slice;
 
 use numaflow_models::models::MonoVertex;
 
-use crate::config::components::{sink, source};
 use crate::config::components::metrics::MetricsConfig;
 use crate::config::components::sink::SinkConfig;
 use crate::config::components::source::{GeneratorConfig, SourceConfig};
 use crate::config::components::transformer::{
     TransformerConfig, TransformerType, UserDefinedConfig,
 };
+use crate::config::components::{sink, source};
 use crate::error::Error;
 use crate::message::get_vertex_replica;
 use crate::Result;
@@ -155,8 +155,8 @@ impl MonovertexConfig {
 
 #[cfg(test)]
 mod tests {
-    use base64::Engine;
     use base64::prelude::BASE64_STANDARD;
+    use base64::Engine;
 
     use crate::config::components::sink::SinkType;
     use crate::config::components::source::SourceType;

rust/numaflow-core/src/metrics.rs

@@ -4,18 +4,12 @@ use std::net::SocketAddr;
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 
-use crate::config::{config, lookback_window_map};
-use crate::source::SourceHandle;
-use crate::Error;
 use axum::body::Body;
 use axum::extract::State;
 use axum::http::{Response, StatusCode};
 use axum::response::IntoResponse;
 use axum::{routing::get, Router};
 use axum_server::tls_rustls::RustlsConfig;
-use numaflow_pb::clients::sink::sink_client::SinkClient;
-use numaflow_pb::clients::source::source_client::SourceClient;
-use numaflow_pb::clients::sourcetransformer::source_transform_client::SourceTransformClient;
 use prometheus_client::encoding::text::encode;
 use prometheus_client::metrics::counter::Counter;
 use prometheus_client::metrics::family::Family;
@@ -30,6 +24,14 @@ use tonic::transport::Channel;
 use tonic::Request;
 use tracing::{debug, error, info};
 
+use numaflow_pb::clients::sink::sink_client::SinkClient;
+use numaflow_pb::clients::source::source_client::SourceClient;
+use numaflow_pb::clients::sourcetransformer::source_transform_client::SourceTransformClient;
+
+use crate::config::lookback_window_map;
+use crate::source::SourceHandle;
+use crate::Error;
+
 pub const COMPONENT_MVTX: &str = "mono-vertex";
 
 // SDK information
@@ -800,13 +802,14 @@ mod tests {
     use std::net::SocketAddr;
     use std::time::Instant;
 
-    use super::*;
-    use crate::config::LookbackStruct;
-    use crate::shared::utils::create_rpc_channel;
     use numaflow::source::{Message, Offset, SourceReadRequest};
     use numaflow::{sink, source, sourcetransform};
     use tokio::sync::mpsc::Sender;
 
+    use crate::shared::utils::create_rpc_channel;
+
+    use super::*;
+
     struct SimpleSource;
     #[tonic::async_trait]
     impl source::Sourcer for SimpleSource {

rust/numaflow-core/src/pipeline.rs

@@ -430,6 +430,7 @@ mod tests {
                 lag_check_interval_in_secs: 5,
                 lag_refresh_interval_in_secs: 3,
             },
+            lookback_seconds: 120,
         };
 
         let cancellation_token = tokio_util::sync::CancellationToken::new();
@@ -585,6 +586,7 @@ mod tests {
                 lag_check_interval_in_secs: 5,
                 lag_refresh_interval_in_secs: 3,
             },
+            lookback_seconds: 120,
         };
 
         let cancellation_token = tokio_util::sync::CancellationToken::new();