feat: stop the procedure manager if a new leader is elected

src/cmd/src/error.rs

@@ -37,6 +37,18 @@ pub enum Error {
         source: common_meta::error::Error,
     },
 
+    #[snafu(display("Failed to start procedure manager"))]
+    StartProcedureManager {
+        location: Location,
+        source: common_procedure::error::Error,
+    },
+
+    #[snafu(display("Failed to stop procedure manager"))]
+    StopProcedureManager {
+        location: Location,
+        source: common_procedure::error::Error,
+    },
+
     #[snafu(display("Failed to start datanode"))]
     StartDatanode {
         location: Location,
@@ -241,7 +253,8 @@ impl ErrorExt for Error {
             | Error::CreateDir { .. }
             | Error::EmptyResult { .. }
             | Error::InvalidDatabaseName { .. } => StatusCode::InvalidArguments,
-
+            Error::StartProcedureManager { source, .. }
+            | Error::StopProcedureManager { source, .. } => source.status_code(),
             Error::ReplCreation { .. } | Error::Readline { .. } => StatusCode::Internal,
             Error::RequestDatabase { source, .. } => source.status_code(),
             Error::CollectRecordBatches { source, .. }

src/cmd/src/standalone.rs

@@ -43,7 +43,8 @@ use snafu::ResultExt;
 
 use crate::error::{
     CreateDirSnafu, IllegalConfigSnafu, InitMetadataSnafu, Result, ShutdownDatanodeSnafu,
-    ShutdownFrontendSnafu, StartDatanodeSnafu, StartFrontendSnafu,
+    ShutdownFrontendSnafu, StartDatanodeSnafu, StartFrontendSnafu, StartProcedureManagerSnafu,
+    StopProcedureManagerSnafu,
 };
 use crate::options::{MixOptions, Options, TopLevelOptions};
 
@@ -163,6 +164,7 @@ impl StandaloneOptions {
 pub struct Instance {
     datanode: Datanode,
     frontend: FeInstance,
+    procedure_manager: ProcedureManagerRef,
 }
 
 impl Instance {
@@ -171,6 +173,11 @@ impl Instance {
         self.datanode.start().await.context(StartDatanodeSnafu)?;
         info!("Datanode instance started");
 
+        self.procedure_manager
+            .start()
+            .await
+            .context(StartProcedureManagerSnafu)?;
+
         self.frontend.start().await.context(StartFrontendSnafu)?;
         Ok(())
     }
@@ -181,6 +188,11 @@ impl Instance {
             .await
             .context(ShutdownFrontendSnafu)?;
 
+        self.procedure_manager
+            .stop()
+            .await
+            .context(StopProcedureManagerSnafu)?;
+
         self.datanode
             .shutdown()
             .await
@@ -354,7 +366,7 @@ impl StartCommand {
         let mut frontend = build_frontend(
             fe_plugins,
             kv_store,
-            procedure_manager,
+            procedure_manager.clone(),
             catalog_manager,
             region_server,
         )
@@ -365,7 +377,11 @@ impl StartCommand {
             .await
             .context(StartFrontendSnafu)?;
 
-        Ok(Instance { datanode, frontend })
+        Ok(Instance {
+            datanode,
+            frontend,
+            procedure_manager,
+        })
     }
 }
 

src/common/meta/src/key.rs

@@ -258,7 +258,6 @@ impl<T: Serialize + DeserializeOwned> DeserializedValueWithBytes<T> {
         self.bytes.to_vec()
     }
 
-    #[cfg(feature = "testing")]
     /// Notes: used for test purpose.
     pub fn from_inner(inner: T) -> Self {
         let bytes = serde_json::to_vec(&inner).unwrap();

src/common/procedure/src/error.rs

@@ -34,6 +34,9 @@ pub enum Error {
     #[snafu(display("Loader {} is already registered", name))]
     LoaderConflict { name: String, location: Location },
 
+    #[snafu(display("Procedure Manager is stopped"))]
+    ManagerNotStart { location: Location },
+
     #[snafu(display("Failed to serialize to json"))]
     ToJson {
         #[snafu(source)]
@@ -148,7 +151,8 @@ impl ErrorExt for Error {
             | Error::FromJson { .. }
             | Error::RetryTimesExceeded { .. }
             | Error::RetryLater { .. }
-            | Error::WaitWatcher { .. } => StatusCode::Internal,
+            | Error::WaitWatcher { .. }
+            | Error::ManagerNotStart { .. } => StatusCode::Internal,
             Error::LoaderConflict { .. } | Error::DuplicateProcedure { .. } => {
                 StatusCode::InvalidArguments
             }

src/common/procedure/src/lib.rs

@@ -14,6 +14,8 @@
 
 //! Common traits and structures for the procedure framework.
 
+#![feature(assert_matches)]
+
 pub mod error;
 pub mod local;
 pub mod options;