diff --git a/src/meta-srv/src/error.rs b/src/meta-srv/src/error.rs index f9de94fc8cf..88ae7142ea2 100644 --- a/src/meta-srv/src/error.rs +++ b/src/meta-srv/src/error.rs @@ -69,6 +69,18 @@ pub enum Error { region_id: RegionId, }, + #[snafu(display( + "The region migration procedure is completed for region: {}, target_peer: {}", + region_id, + target_peer_id + ))] + RegionMigrated { + #[snafu(implicit)] + location: Location, + region_id: RegionId, + target_peer_id: u64, + }, + #[snafu(display("The region migration procedure aborted, reason: {}", reason))] MigrationAbort { #[snafu(implicit)] @@ -953,7 +965,8 @@ impl ErrorExt for Error { | Error::RegionOpeningRace { .. } | Error::RegionRouteNotFound { .. } | Error::MigrationAbort { .. } - | Error::MigrationRunning { .. } => StatusCode::Unexpected, + | Error::MigrationRunning { .. } + | Error::RegionMigrated { .. } => StatusCode::Unexpected, Error::TableNotFound { .. } => StatusCode::TableNotFound, Error::SaveClusterInfo { source, .. } | Error::InvalidClusterInfoFormat { source, .. } diff --git a/src/meta-srv/src/procedure/region_migration/manager.rs b/src/meta-srv/src/procedure/region_migration/manager.rs index 9286bb87784..b277bd3e23b 100644 --- a/src/meta-srv/src/procedure/region_migration/manager.rs +++ b/src/meta-srv/src/procedure/region_migration/manager.rs @@ -326,7 +326,11 @@ impl RegionMigrationManager { if self.has_migrated(®ion_route, &task)? { info!("Skipping region migration task: {task}"); - return Ok(None); + return error::RegionMigratedSnafu { + region_id, + target_peer_id: task.to_peer.id, + } + .fail(); } self.verify_region_leader_peer(®ion_route, &task)?; @@ -567,7 +571,8 @@ mod test { env.create_physical_table_metadata(table_info, region_routes) .await; - manager.submit_procedure(task).await.unwrap(); + let err = manager.submit_procedure(task).await.unwrap_err(); + assert_matches!(err, error::Error::RegionMigrated { .. }); } #[tokio::test] diff --git a/src/meta-srv/src/region/supervisor.rs b/src/meta-srv/src/region/supervisor.rs index 29867207cc4..edbe29d85ca 100644 --- a/src/meta-srv/src/region/supervisor.rs +++ b/src/meta-srv/src/region/supervisor.rs @@ -27,7 +27,7 @@ use common_meta::DatanodeId; use common_runtime::JoinHandle; use common_telemetry::{debug, error, info, warn}; use common_time::util::current_time_millis; -use error::Error::{LeaderPeerChanged, MigrationRunning, TableRouteNotFound}; +use error::Error::{LeaderPeerChanged, MigrationRunning, RegionMigrated, TableRouteNotFound}; use snafu::{ensure, OptionExt, ResultExt}; use store_api::storage::RegionId; use tokio::sync::mpsc::{Receiver, Sender}; @@ -527,6 +527,7 @@ impl RegionSupervisor { async fn do_failover(&mut self, task: RegionMigrationProcedureTask, count: u32) -> Result<()> { let from_peer_id = task.from_peer.id; + let to_peer_id = task.to_peer.id; let region_id = task.region_id; info!( @@ -536,6 +537,15 @@ impl RegionSupervisor { if let Err(err) = self.region_migration_manager.submit_procedure(task).await { return match err { + RegionMigrated { .. } => { + info!( + "Region has been migrated to target peer: {}, removed failover detector for region: {}, datanode: {}", + to_peer_id, region_id, from_peer_id + ); + self.deregister_failure_detectors(vec![(from_peer_id, region_id)]) + .await; + Ok(()) + } // Returns Ok if it's running or table is dropped. MigrationRunning { .. } => { info!( diff --git a/tests-integration/tests/region_migration.rs b/tests-integration/tests/region_migration.rs index 28d6371c8bc..553160ac423 100644 --- a/tests-integration/tests/region_migration.rs +++ b/tests-integration/tests/region_migration.rs @@ -32,6 +32,7 @@ use datatypes::vectors::{Helper, UInt64Vector}; use frontend::error::Result as FrontendResult; use frontend::instance::Instance; use futures::future::BoxFuture; +use meta_srv::error; use meta_srv::error::Result as MetaResult; use meta_srv::metasrv::SelectorContext; use meta_srv::procedure::region_migration::RegionMigrationProcedureTask; @@ -213,7 +214,7 @@ pub async fn test_region_migration(store_type: StorageType, endpoints: Vec