Fix panic when region is not found on TiFlash's side in tryFlushData …

…to release-6.2 (#5606)

close #5602

dbms/src/Storages/Transaction/KVStore.cpp

@@ -331,21 +331,31 @@ bool KVStore::needFlushRegionData(UInt64 region_id, TMTContext & tmt)
 {
     auto region_task_lock = region_manager.genRegionTaskLock(region_id);
     const RegionPtr curr_region_ptr = getRegion(region_id);
+    // TODO Should handle when curr_region_ptr is null.
     return canFlushRegionDataImpl(curr_region_ptr, false, false, tmt, region_task_lock, 0, 0);
 }
 
 bool KVStore::tryFlushRegionData(UInt64 region_id, bool try_until_succeed, TMTContext & tmt, UInt64 index, UInt64 term)
 {
     auto region_task_lock = region_manager.genRegionTaskLock(region_id);
     const RegionPtr curr_region_ptr = getRegion(region_id);
+    if (curr_region_ptr == nullptr)
+    {
+        /// If we can't find region here, we return true so proxy can trigger a CompactLog.
+        /// The triggered CompactLog will be handled by `handleUselessAdminRaftCmd`,
+        /// and result in a `EngineStoreApplyRes::NotFound`.
+        /// Proxy will print this message and continue: `region not found in engine-store, maybe have exec `RemoveNode` first`.
+        LOG_FMT_WARNING(log, "region {} [index: {}, term {}], not exist when flushing, maybe have exec `RemoveNode` first", region_id, index, term);
+        return true;
+    }
     return canFlushRegionDataImpl(curr_region_ptr, true, try_until_succeed, tmt, region_task_lock, index, term);
 }
 
 bool KVStore::canFlushRegionDataImpl(const RegionPtr & curr_region_ptr, UInt8 flush_if_possible, bool try_until_succeed, TMTContext & tmt, const RegionTaskLock & region_task_lock, UInt64 index, UInt64 term)
 {
     if (curr_region_ptr == nullptr)
     {
-        throw Exception(fmt::format("region not found when trying flush", ErrorCodes::LOGICAL_ERROR));
+        throw Exception("region not found when trying flush", ErrorCodes::LOGICAL_ERROR);
     }
     auto & curr_region = *curr_region_ptr;
 

dbms/src/Storages/Transaction/tests/gtest_kvstore.cpp

@@ -1242,6 +1242,14 @@ void RegionKVStoreTest::testKVStore()
         ASSERT_EQ(kvs.needFlushRegionData(19, ctx.getTMTContext()), true);
         // Force flush until succeed only for testing.
         ASSERT_EQ(kvs.tryFlushRegionData(19, true, ctx.getTMTContext(), 0, 0), true);
+        // Non existing region.
+        // Flush and CompactLog will not panic.
+        ASSERT_EQ(kvs.tryFlushRegionData(1999, true, ctx.getTMTContext(), 0, 0), true);
+        raft_cmdpb::AdminRequest request;
+        raft_cmdpb::AdminResponse response;
+        request.mutable_compact_log();
+        request.set_cmd_type(::raft_cmdpb::AdminCmdType::CompactLog);
+        ASSERT_EQ(kvs.handleAdminRaftCmd(raft_cmdpb::AdminRequest{request}, std::move(response), 1999, 22, 6, ctx.getTMTContext()), EngineStoreApplyRes::NotFound);
     }
 }