feat: flow refill time window after reboot

src/cmd/src/flownode.rs

@@ -21,6 +21,7 @@ use clap::Parser;
 use client::client_manager::NodeClients;
 use common_base::Plugins;
 use common_config::Configurable;
+use common_error::ext::BoxedError;
 use common_grpc::channel_manager::ChannelConfig;
 use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
 use common_meta::heartbeat::handler::invalidate_table_cache::InvalidateCacheHandler;
@@ -38,8 +39,8 @@ use snafu::{OptionExt, ResultExt};
 use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{
-    BuildCacheRegistrySnafu, InitMetadataSnafu, LoadLayeredConfigSnafu, MetaClientInitSnafu,
-    MissingConfigSnafu, Result, ShutdownFlownodeSnafu, StartFlownodeSnafu,
+    BuildCacheRegistrySnafu, BuildCliSnafu, InitMetadataSnafu, LoadLayeredConfigSnafu,
+    MetaClientInitSnafu, MissingConfigSnafu, Result, ShutdownFlownodeSnafu, StartFlownodeSnafu,
 };
 use crate::options::{GlobalOptions, GreptimeOptions};
 use crate::{log_versions, App};
@@ -301,7 +302,7 @@ impl StartCommand {
             Plugins::new(),
             table_metadata_manager,
             catalog_manager.clone(),
-            flow_metadata_manager,
+            flow_metadata_manager.clone(),
         )
         .with_heartbeat_task(heartbeat_task);
 
@@ -316,7 +317,7 @@ impl StartCommand {
         let client = Arc::new(NodeClients::new(channel_config));
 
         let invoker = FrontendInvoker::build_from(
-            flownode.flow_worker_manager().clone(),
+            None,
             catalog_manager.clone(),
             cached_meta_backend.clone(),
             layered_cache_registry.clone(),
@@ -330,6 +331,16 @@ impl StartCommand {
             .set_frontend_invoker(invoker)
             .await;
 
+        if let Err(err) = flownode
+            .flow_worker_manager()
+            .create_and_start_refill_flow_tasks(&flow_metadata_manager, &(catalog_manager as _))
+            .await
+            .map_err(BoxedError::new)
+            .context(BuildCliSnafu)
+        {
+            common_telemetry::error!(?err, "Failed to create and start refill flow tasks");
+        }
+
         Ok(Instance::new(flownode, guard))
     }
 }

src/cmd/src/standalone.rs

@@ -76,10 +76,10 @@ use tokio::sync::{broadcast, RwLock};
 use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{
-    BuildCacheRegistrySnafu, CreateDirSnafu, IllegalConfigSnafu, InitDdlManagerSnafu,
-    InitMetadataSnafu, InitTimezoneSnafu, LoadLayeredConfigSnafu, OtherSnafu, Result,
-    ShutdownDatanodeSnafu, ShutdownFlownodeSnafu, ShutdownFrontendSnafu, StartDatanodeSnafu,
-    StartFlownodeSnafu, StartFrontendSnafu, StartProcedureManagerSnafu,
+    BuildCacheRegistrySnafu, BuildCliSnafu, CreateDirSnafu, IllegalConfigSnafu,
+    InitDdlManagerSnafu, InitMetadataSnafu, InitTimezoneSnafu, LoadLayeredConfigSnafu, OtherSnafu,
+    Result, ShutdownDatanodeSnafu, ShutdownFlownodeSnafu, ShutdownFrontendSnafu,
+    StartDatanodeSnafu, StartFlownodeSnafu, StartFrontendSnafu, StartProcedureManagerSnafu,
     StartWalOptionsAllocatorSnafu, StopProcedureManagerSnafu,
 };
 use crate::options::{GlobalOptions, GreptimeOptions};
@@ -580,7 +580,7 @@ impl StartCommand {
             layered_cache_registry.clone(),
             table_metadata_manager,
             table_meta_allocator,
-            flow_metadata_manager,
+            flow_metadata_manager.clone(),
             flow_meta_allocator,
         )
         .await?;
@@ -602,7 +602,7 @@ impl StartCommand {
         let flow_worker_manager = flownode.flow_worker_manager();
         // flow server need to be able to use frontend to write insert requests back
         let invoker = FrontendInvoker::build_from(
-            flow_worker_manager.clone(),
+            Some(frontend.query_engine()),
             catalog_manager.clone(),
             kv_backend.clone(),
             layered_cache_registry.clone(),
@@ -613,6 +613,15 @@ impl StartCommand {
         .context(StartFlownodeSnafu)?;
         flow_worker_manager.set_frontend_invoker(invoker).await;
 
+        if let Err(err) = flow_worker_manager
+            .create_and_start_refill_flow_tasks(&flow_metadata_manager, &(catalog_manager as _))
+            .await
+            .map_err(BoxedError::new)
+            .context(BuildCliSnafu)
+        {
+            common_telemetry::error!(err; "failed to refill flow");
+        }
+
         let (tx, _rx) = broadcast::channel(1);
 
         let servers = Services::new(opts, Arc::new(frontend.clone()), plugins)