fix: sync failure detectors during soft-drop lifecycle

Keep region failure detector state aligned as soft-dropped tables close, reopen, and purge regions so stale detectors do not trigger failover for unavailable or deleted regions.

Files:

- \`src/common/meta/src/ddl.rs\`

- \`src/common/meta/src/ddl/drop_table.rs\`

- \`src/common/meta/src/ddl/undrop_table.rs\`

- \`src/common/meta/src/ddl/purge_dropped_table.rs\`

- \`src/common/meta/src/ddl/tests/drop_table.rs\`

Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
This commit is contained in:
Lei, HUANG
2026-07-02 14:35:52 +08:00
committed by Lei, HUANG
parent 1df0c58269
commit 7f6fd9e61f
5 changed files with 190 additions and 7 deletions
+4 -1
View File
@@ -138,7 +138,10 @@ impl DdlContext {
///
/// Once the regions were dropped, subsequent heartbeats no longer include these regions.
/// Therefore, we should remove the failure detectors for these dropped regions.
async fn deregister_failure_detectors(&self, detecting_regions: Vec<DetectingRegion>) {
pub(crate) async fn deregister_failure_detectors(
&self,
detecting_regions: Vec<DetectingRegion>,
) {
self.region_failure_detector_controller
.deregister_failure_detectors(detecting_regions)
.await;
+6 -1
View File
@@ -36,7 +36,7 @@ use table::table_reference::TableReference;
use self::executor::DropTableExecutor;
use crate::ddl::DdlContext;
use crate::ddl::utils::map_to_procedure_error;
use crate::ddl::utils::{convert_region_routes_to_detecting_regions, map_to_procedure_error};
use crate::error::{self, Result};
use crate::key::table_route::TableRouteValue;
use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
@@ -181,6 +181,11 @@ impl DropTableProcedure {
&self.data.physical_region_routes,
)
.await?;
self.context
.deregister_failure_detectors(convert_region_routes_to_detecting_regions(
&self.data.physical_region_routes,
))
.await;
self.dropping_regions.clear();
return Ok(Status::done());
}
@@ -30,7 +30,10 @@ use table::table_name::TableName;
use crate::ddl::DdlContext;
use crate::ddl::drop_table::executor::DropTableExecutor;
use crate::ddl::undrop_table::open_regions;
use crate::ddl::utils::{is_metric_engine_logical_table, map_to_procedure_error};
use crate::ddl::utils::{
convert_region_routes_to_detecting_regions, is_metric_engine_logical_table,
map_to_procedure_error,
};
use crate::error::{self, Result};
use crate::key::table_route::TableRouteValue;
use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
@@ -119,6 +122,11 @@ impl PurgeDroppedTableProcedure {
false,
)
.await?;
self.context
.deregister_failure_detectors(convert_region_routes_to_detecting_regions(
region_routes,
))
.await;
}
self.data.state = PurgeDroppedTableState::DeleteTombstone;
Ok(Status::executing(true))
+166 -2
View File
@@ -17,6 +17,7 @@ use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use api::v1::region::{RegionRequest, region_request};
use async_trait::async_trait;
use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
use common_error::ext::ErrorExt;
use common_error::status_code::StatusCode;
@@ -27,9 +28,8 @@ use common_procedure_test::{
use store_api::region_engine::RegionRole;
use store_api::storage::RegionId;
use table::metadata::TableId;
use tokio::sync::mpsc;
use tokio::sync::{Mutex, mpsc};
use crate::ddl::TableMetadata;
use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
use crate::ddl::create_table::CreateTableProcedure;
use crate::ddl::drop_table::{DropTableProcedure, DropTableState};
@@ -41,6 +41,7 @@ use crate::ddl::test_util::{
put_datanode_address, test_create_logical_table_task, test_create_physical_table_task,
};
use crate::ddl::undrop_table::UndropTableProcedure;
use crate::ddl::{DetectingRegion, RegionFailureDetectorController, TableMetadata};
use crate::error::Error;
use crate::key::table_name::TableNameKey;
use crate::key::table_route::TableRouteValue;
@@ -319,6 +320,45 @@ async fn test_soft_drop_closes_regions_and_keeps_tombstone() {
assert!(dropped_table.is_some());
}
#[tokio::test]
async fn test_soft_drop_deregisters_failure_detectors() {
let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default());
let mut ddl_context = new_ddl_context(node_manager);
ddl_context.soft_drop_enabled = true;
ddl_context.region_failure_detector_controller = detector_controller.clone();
let table_id = 1024;
let table_name = "foo";
let task = test_create_table_task(table_name, table_id);
ddl_context
.table_metadata_manager
.create_table_metadata(
task.table_info.clone(),
TableRouteValue::physical(vec![RegionRoute {
region: Region::new_test(RegionId::new(table_id, 1)),
leader_peer: Some(Peer::empty(1)),
follower_peers: vec![Peer::empty(2)],
leader_state: None,
leader_down_since: None,
write_route_policy: None,
}]),
HashMap::new(),
)
.await
.unwrap();
let mut procedure = DropTableProcedure::new(
new_drop_table_task(table_name, table_id, false),
ddl_context,
);
execute_procedure_until_done(&mut procedure).await;
assert_eq!(
detector_controller.deregistered().await,
vec![(1, RegionId::new(table_id, 1))]
);
}
#[tokio::test]
async fn test_hard_drop_keeps_delete_tombstone_flow() {
let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
@@ -568,6 +608,49 @@ async fn test_undrop_table_restores_metadata_and_reopens_regions() {
assert!(rx.try_recv().is_err());
}
#[tokio::test]
async fn test_undrop_table_registers_failure_detectors() {
let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default());
let mut ddl_context = new_ddl_context(node_manager);
ddl_context.soft_drop_enabled = true;
ddl_context.region_failure_detector_controller = detector_controller.clone();
let table_id = 1024;
let table_name = "foo";
let task = test_create_table_task(table_name, table_id);
ddl_context
.table_metadata_manager
.create_table_metadata(
task.table_info.clone(),
TableRouteValue::physical(vec![RegionRoute {
region: Region::new_test(RegionId::new(table_id, 1)),
leader_peer: Some(Peer::empty(1)),
follower_peers: vec![],
leader_state: None,
leader_down_since: None,
write_route_policy: None,
}]),
HashMap::new(),
)
.await
.unwrap();
let mut drop_procedure = DropTableProcedure::new(
new_drop_table_task(table_name, table_id, false),
ddl_context.clone(),
);
execute_procedure_until_done(&mut drop_procedure).await;
detector_controller.clear().await;
let mut procedure =
UndropTableProcedure::new(new_undrop_table_task(table_name, table_id), ddl_context);
execute_procedure_until_done(&mut procedure).await;
assert_eq!(
detector_controller.registered().await,
vec![(1, RegionId::new(table_id, 1))]
);
}
#[tokio::test]
async fn test_undrop_logical_table_skips_datanode_open() {
let (tx, mut rx) = mpsc::channel(8);
@@ -787,6 +870,55 @@ async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() {
);
}
#[tokio::test]
async fn test_purge_dropped_table_deregisters_failure_detectors() {
let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default());
let mut ddl_context = new_ddl_context(node_manager);
ddl_context.region_failure_detector_controller = detector_controller.clone();
let table_id = 1024;
let table_name = "foo";
let task = test_create_table_task(table_name, table_id);
let table_route_value = TableRouteValue::physical(vec![RegionRoute {
region: Region::new_test(RegionId::new(table_id, 1)),
leader_peer: Some(Peer::empty(1)),
follower_peers: vec![],
leader_state: None,
leader_down_since: None,
write_route_policy: None,
}]);
ddl_context
.table_metadata_manager
.create_table_metadata(
task.table_info.clone(),
table_route_value.clone(),
HashMap::new(),
)
.await
.unwrap();
ddl_context
.table_metadata_manager
.delete_table_metadata(
table_id,
&task.table_name(),
&table_route_value,
&HashMap::new(),
)
.await
.unwrap();
let mut procedure = PurgeDroppedTableProcedure::new(
new_purge_dropped_table_task(table_name, Some(table_id)),
ddl_context,
);
execute_procedure_until_done(&mut procedure).await;
assert_eq!(
detector_controller.deregistered().await,
vec![(1, RegionId::new(table_id, 1))]
);
}
#[tokio::test]
async fn test_purge_dropped_table_by_name_selects_tombstone_when_live_table_exists() {
let (tx, mut rx) = mpsc::channel(8);
@@ -976,6 +1108,38 @@ async fn create_metric_logical_table_tombstone(
logical_table_id
}
#[derive(Default)]
struct RecordingRegionFailureDetectorController {
registered: Mutex<Vec<DetectingRegion>>,
deregistered: Mutex<Vec<DetectingRegion>>,
}
impl RecordingRegionFailureDetectorController {
async fn registered(&self) -> Vec<DetectingRegion> {
self.registered.lock().await.clone()
}
async fn deregistered(&self) -> Vec<DetectingRegion> {
self.deregistered.lock().await.clone()
}
async fn clear(&self) {
self.registered.lock().await.clear();
self.deregistered.lock().await.clear();
}
}
#[async_trait]
impl RegionFailureDetectorController for RecordingRegionFailureDetectorController {
async fn register_failure_detectors(&self, detecting_regions: Vec<DetectingRegion>) {
self.registered.lock().await.extend(detecting_regions);
}
async fn deregister_failure_detectors(&self, detecting_regions: Vec<DetectingRegion>) {
self.deregistered.lock().await.extend(detecting_regions);
}
}
#[tokio::test]
async fn test_memory_region_keeper_guard_dropped_on_procedure_done() {
let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
+5 -2
View File
@@ -34,8 +34,8 @@ use table::table_name::TableName;
use table::table_reference::TableReference;
use crate::ddl::utils::{
add_peer_context_if_needed, is_metric_engine_logical_table, map_to_procedure_error,
region_storage_path,
add_peer_context_if_needed, convert_region_routes_to_detecting_regions,
is_metric_engine_logical_table, map_to_procedure_error, region_storage_path,
};
use crate::ddl::{CreateRequestBuilder, DdlContext, build_template_from_raw_table_info};
use crate::error::{self, Result};
@@ -258,6 +258,9 @@ pub(crate) async fn open_regions(
.await
.into_iter()
.collect::<Result<Vec<_>>>()?;
context
.register_failure_detectors(convert_region_routes_to_detecting_regions(region_routes))
.await;
Ok(())
}