diff --git a/src/log-store/src/error.rs b/src/log-store/src/error.rs index 962606666b..286f3b5fb6 100644 --- a/src/log-store/src/error.rs +++ b/src/log-store/src/error.rs @@ -17,6 +17,7 @@ use std::any::Any; use common_error::ext::ErrorExt; use common_macro::stack_trace_debug; use common_runtime::error::Error as RuntimeError; +use common_runtime::JoinError; use serde_json::error::Error as JsonError; use snafu::{Location, Snafu}; use store_api::storage::RegionId; @@ -306,6 +307,14 @@ pub enum Error { #[snafu(implicit)] location: Location, }, + + #[snafu(display("Join error"))] + Join { + #[snafu(implicit)] + location: Location, + #[snafu(source)] + error: JoinError, + }, } impl ErrorExt for Error { diff --git a/src/log-store/src/raft_engine/log_store.rs b/src/log-store/src/raft_engine/log_store.rs index a4db95cd57..419621e578 100644 --- a/src/log-store/src/raft_engine/log_store.rs +++ b/src/log-store/src/raft_engine/log_store.rs @@ -31,8 +31,8 @@ use store_api::storage::RegionId; use crate::error::{ AddEntryLogBatchSnafu, DiscontinuousLogIndexSnafu, Error, FetchEntrySnafu, - IllegalNamespaceSnafu, IllegalStateSnafu, InvalidProviderSnafu, OverrideCompactedEntrySnafu, - RaftEngineSnafu, Result, StartGcTaskSnafu, StopGcTaskSnafu, + IllegalNamespaceSnafu, IllegalStateSnafu, InvalidProviderSnafu, JoinSnafu, + OverrideCompactedEntrySnafu, RaftEngineSnafu, Result, StartGcTaskSnafu, StopGcTaskSnafu, }; use crate::metrics; use crate::raft_engine::backend::SYSTEM_NAMESPACE; @@ -250,6 +250,12 @@ impl LogStore for RaftEngineLogStore { .engine .write(&mut batch, sync) .context(RaftEngineSnafu)?; + let engine = self.engine.clone(); + let _ = common_runtime::spawn_blocking_global(move || { + engine.write(&mut batch, sync).context(RaftEngineSnafu) + }) + .await + .context(JoinSnafu)?; Ok(AppendBatchResponse { last_entry_ids }) }