Repository: tikv/tikv (language: rust)

Issue:
GC worker still uses ConcurrencyManager::new which leads to annoying error logs
## Bug Report

<!-- Thanks for your bug report! Don't worry if you can't fill out all the sections. -->

### What version of TiKV are you using?
<!-- You can run `tikv-server --version` -->
v8.5.1

### What operating system and CPU are you using?
<!-- If you're using Linux, you can run `cat /proc/cpuinfo` -->
unreleated

### Steps to reproduce
<!-- If possible, provide a recipe for reproducing the error. A complete runnable program is good. -->
run tikv-server

### What did you expect?

No annoying error message

### What did happened?

TiKV prints a lot of following logs because GC worker still uses ConcurrencyManager::new

```
[2025/02/12 14:15:22.320 +08:00] [ERROR] [lib.rs:133 ["improper setting: limit_valid_duration >= max_ts_drift_allowance; consider increasing max-ts-drift-allowance or decreasing max-ts-sync-interval"] [max_ts_drift_allowance=0ns] [limit_valid_duration=45s] [thread_id=32]
[2025/02/12 14:15:22.321 +08:00] [ERROR] [lib.rs:133 ["improper setting: limit_valid_duration >= max_ts_drift_allowance; consider increasing max-ts-drift-allowance or decreasing max-ts-sync-interval"] [max_ts_drift_allowance=0ns] [limit_valid_duration=45s] [thread_id=32]
```


The backtrace
```
backtrace="   0: tikv_util::set_panic_hook::{{closure}}
             at /home/ywxk/src/tikv_master/components/tikv_util/src/lib.rs:480:18
   1: <alloc::boxed::Box<F,A> as core::ops::function::Fn<Args>>::call
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2029:9
   2: std::panicking::rust_panic_with_hook
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:783:13
   3: std::panicking::begin_panic_handler::{{closure}}
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:649:13
   4: std::sys_common::backtrace::__rust_end_short_backtrace
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:171:18
   5: rust_begin_unwind
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:645:5
   6: core::panicking::panic_fmt
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:72:14
   7: concurrency_manager::ConcurrencyManager::new_with_config
             at /home/ywxk/src/tikv_master/components/concurrency_manager/src/lib.rs:136:13
   8: concurrency_manager::ConcurrencyManager::new
             at /home/ywxk/src/tikv_master/components/concurrency_manager/src/lib.rs:109:9
   9: tikv::server::gc_worker::gc_worker::GcRunnerCore<E>::new_txn
             at /home/ywxk/src/tikv_master/src/server/gc_worker/gc_worker.rs:371:35
  10: tikv::server::gc_worker::gc_worker::GcRunnerCore<E>::gc_keys
             at /home/ywxk/src/tikv_master/src/server/gc_worker/gc_worker.rs:466:27
  11: tikv::server::gc_worker::gc_worker::GcRunnerCore<E>::gc
             at /home/ywxk/src/tikv_master/src/server/gc_worker/gc_worker.rs:417:13
  12: tikv::server::gc_worker::gc_worker::GcRunnerCore<E>::run
             at /home/ywxk/src/tikv_master/src/server/gc_worker/gc_worker.rs:973:27
  13: <tikv::server::gc_worker::gc_worker::GcRunner<E> as tikv_util::worker::pool::Runnable>::run::{{closure}}
             at /home/ywxk/src/tikv_master/src/server/gc_worker/gc_worker.rs:1132:13
  14: yatp::task::future::RawTask<F>::poll
             at /home/ywxk/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:59:9
  15: yatp::task::future::TaskCell::poll
             at /home/ywxk/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:103:9
  16: <yatp::task::future::Runner as yatp::pool::runner::Runner>::handle
             at /home/ywxk/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:387:20
  17: <tikv_util::yatp_pool::YatpPoolRunner<T> as yatp::pool::runner::Runner>::handle
             at /home/ywxk/src/tikv_master/components/tikv_util/src/yatp_pool/mod.rs:199:24
  18: yatp::pool::worker::WorkerThread<T,R>::run
             at /home/ywxk/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/worker.rs:48:13
  19: yatp::pool::builder::LazyBuilder<T>::build::{{closure}}
             at /home/ywxk/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/builder.rs:114:25
  20: std::sys_common::backtrace::__rust_begin_short_backtrace
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:155:18
  21: std::thread::Builder::spawn_unchecked_::{{closure}}::{{closure}}
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:529:17
  22: <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panic/unwind_safe.rs:272:9
  23: std::panicking::try::do_call
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:552:40
  24: __rust_try
  25: std::panicking::try
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:516:19
  26: std::panic::catch_unwind
             at /home/ywxk/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panic.rs:142:14
```

When `compaction filter is disabled` by setting 
```
[gc]
enable-compaction-filter = false
```
it is likely to happen.


Discussion / hints:

Fix patch:
diff --git a/components/backup/src/endpoint.rs b/components/backup/src/endpoint.rs
index 0f93ac109ce..8bd082b4b98 100644
--- a/components/backup/src/endpoint.rs
+++ b/components/backup/src/endpoint.rs
@@ -1479,7 +1479,7 @@ pub mod tests {
             .api_version(api_version)
             .build()
             .unwrap();
-        let concurrency_manager = ConcurrencyManager::new(1.into());
+        let concurrency_manager = ConcurrencyManager::new_for_test(1.into());
         let need_encode_key = !is_raw_kv || api_version == ApiVersion::V2;
         let db = rocks.get_rocksdb();
         (
diff --git a/components/cdc/src/endpoint.rs b/components/cdc/src/endpoint.rs
index da33bfd4021..7bcfb5a1085 100644
--- a/components/cdc/src/endpoint.rs
+++ b/components/cdc/src/endpoint.rs
@@ -1512,7 +1512,7 @@ mod tests {
             })),
             CdcObserver::new(task_sched, memory_quota.clone()),
             Arc::new(StdMutex::new(store_meta)),
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             env,
             security_mgr,
             memory_quota,
diff --git a/components/concurrency_manager/benches/lock_table.rs b/components/concurrency_manager/benches/lock_table.rs
index 33ae220ee51..d7df251d70e 100644
--- a/components/concurrency_manager/benches/lock_table.rs
+++ b/components/concurrency_manager/benches/lock_table.rs
@@ -15,7 +15,7 @@ const KEY_LEN: usize = 64;
 const LOCK_COUNT: usize = 10_000;
 
 fn prepare_cm() -> ConcurrencyManager {
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     let mut buf = [0; KEY_LEN];
     for _ in 0..LOCK_COUNT {
         thread_rng().fill_bytes(&mut buf[..]);
diff --git a/components/concurrency_manager/src/lib.rs b/components/concurrency_manager/src/lib.rs
index e1b76dd213a..c26f6af012d 100644
--- a/components/concurrency_manager/src/lib.rs
+++ b/components/concurrency_manager/src/lib.rs
@@ -105,13 +105,28 @@ pub struct ConcurrencyManager {
 }
 
 impl ConcurrencyManager {
-    pub fn new(latest_ts: TimeStamp) -> Self {
+    /// This is ONLY used in tests, please do not use it elsewhere.
+    /// To create a new concurrency manager, please use `new_with_config`
+    /// instead.
+    pub fn new_for_test(latest_ts: TimeStamp) -> Self {
         Self::new_with_config(
             latest_ts,
             DEFAULT_LIMIT_VALID_DURATION,
             ActionOnInvalidMaxTs::Panic,
             None,
-            Duration::ZERO,
+            DEFAULT_LIMIT_VALID_DURATION + Duration::from_secs(1),
+        )
+    }
+
+    /// This is ONLY used by `GcRunnerCore` as a temporary solution, please do
+    /// not use it elsewhere.
+    pub fn new_dummy() -> Self {
+        Self::new_with_config(
+            1.into(),
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Log,
+            None,
+            DEFAULT_LIMIT_VALID_DURATION + Duration::from_secs(1),
         )
     }
 
@@ -647,7 +662,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_lock_keys_order() {
-        let concurrency_manager = ConcurrencyManager::new(1.into());
+        let concurrency_manager = ConcurrencyManager::new_for_test(1.into());
         let keys: Vec<_> = [b"c", b"a", b"b"]
             .iter()
             .copied()
@@ -661,7 +676,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_update_max_ts() {
-        let concurrency_manager = ConcurrencyManager::new(10.into());
+        let concurrency_manager = ConcurrencyManager::new_for_test(10.into());
         let _ = concurrency_manager.update_max_ts(20.into(), "");
         assert_eq!(concurrency_manager.max_ts(), 20.into());
 
@@ -689,7 +704,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_global_min_lock_ts() {
-        let concurrency_manager = ConcurrencyManager::new(1.into());
+        let concurrency_manager = ConcurrencyManager::new_for_test(1.into());
 
         assert_eq!(concurrency_manager.global_min_lock_ts(), None);
         let guard = concurrency_manager.lock_key(&Key::from_raw(b"a")).await;
@@ -754,20 +769,20 @@ mod tests {
 
     #[test]
     fn test_max_ts_limit_edge_cases() {
-        let cm = ConcurrencyManager::new(TimeStamp::new(100));
+        let cm = ConcurrencyManager::new_for_test(TimeStamp::new(100));
 
         // Test transition from zero limit
         assert_eq!(cm.max_ts_limit.load().limit, 0.into());
         cm.set_max_ts_limit(TimeStamp::new(1000));
-        assert_eq!(cm.max_ts_limit.load().limit, 1000.into());
+        assert_eq!(cm.max_ts_limit.load().limit, 12058625000.into());
 
         // Try to lower from 1000 to 500 - should be ignored
         cm.set_max_ts_limit(TimeStamp::new(500));
-        assert_eq!(cm.max_ts_limit.load().limit, 1000.into());
+        assert_eq!(cm.max_ts_limit.load().limit, 12058625000.into());
 
         // Test setting limit to max, should have no effect
         cm.set_max_ts_limit(TimeStamp::max());
-        assert_eq!(cm.max_ts_limit.load().limit, 1000.into());
+        assert_eq!(cm.max_ts_limit.load().limit, 12058625000.into());
     }
 
     #[test]
@@ -885,7 +900,7 @@ mod tests {
 
     #[test]
     fn test_update_max_ts_without_limit() {
-        let cm = ConcurrencyManager::new(TimeStamp::new(100));
+        let cm = ConcurrencyManager::new_for_test(TimeStamp::new(100));
 
         cm.update_max_ts(TimeStamp::new(500), "test_source".to_string())
             .unwrap();
diff --git a/components/resolved_ts/src/cmd.rs b/components/resolved_ts/src/cmd.rs
index 7a848381269..7f6a36d9184 100644
--- a/components/resolved_ts/src/cmd.rs
+++ b/components/resolved_ts/src/cmd.rs
@@ -471,7 +471,7 @@ mod tests {
         assert_eq!(rows, expected);
 
         let snapshot = engine.snapshot(Default::default()).unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
         let mut txn = MvccTxn::new(10.into(), cm);
         let mut reader = SnapshotReader::new(10.into(), snapshot, true);
         prewrite(
diff --git a/src/coprocessor/endpoint.rs b/src/coprocessor/endpoint.rs
index 770df34d773..02498ec488c 100644
--- a/src/coprocessor/endpoint.rs
+++ b/src/coprocessor/endpoint.rs
@@ -1152,7 +1152,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1194,7 +1194,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1233,7 +1233,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1257,7 +1257,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1306,7 +1306,7 @@ mod tests {
             .collect::<Vec<_>>(),
         );
 
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1358,7 +1358,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1384,7 +1384,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1438,7 +1438,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1467,7 +1467,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -1564,7 +1564,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config {
                 end_point_stream_channel_size: 3,
@@ -1637,7 +1637,7 @@ mod tests {
             ..Default::default()
         };
 
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &config,
             read_pool.handle(),
@@ -2017,7 +2017,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
@@ -2074,7 +2074,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let key = Key::from_raw(b"key");
         let guard = block_on(cm.lock_key(&key));
         guard.with_lock(|lock| {
@@ -2137,7 +2137,7 @@ mod tests {
             &CoprReadPoolConfig::default_for_test(),
             engine,
         ));
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let copr = Endpoint::<RocksEngine>::new(
             &Config::default(),
             read_pool.handle(),
diff --git a/src/server/gc_worker/gc_worker.rs b/src/server/gc_worker/gc_worker.rs
index 56acc7c1594..d1dc58ae325 100644
--- a/src/server/gc_worker/gc_worker.rs
+++ b/src/server/gc_worker/gc_worker.rs
@@ -366,9 +366,12 @@ impl<E: Engine> GcRunnerCore<E> {
         Ok(())
     }
 
-    fn new_txn() -> MvccTxn {
+    /// Creates a `MvccTxn` with a dummy concurrency manager, do not use it
+    /// beside the `GCRunner`. It's better to abstract `MvccTxn` for
+    /// recording modifies only.
+    fn mvcc_txn_with_dummy_cm() -> MvccTxn {
         // TODO txn only used for GC, but this is hacky, maybe need an Option?
-        let concurrency_manager = ConcurrencyManager::new(1.into());
+        let concurrency_manager = ConcurrencyManager::new_dummy();
         MvccTxn::new(TimeStamp::zero(), concurrency_manager)
     }
 
@@ -463,7 +466,7 @@ impl<E: Engine> GcRunnerCore<E> {
         let mut gc_info = GcInfo::default();
         let mut keys = keys.into_iter().peekable();
         for region in regions {
-            let mut txn = Self::new_txn();
+            let mut txn = Self::mvcc_txn_with_dummy_cm();
             let mut reader = self.create_reader(
                 count,
                 &region,
@@ -516,7 +519,7 @@ impl<E: Engine> GcRunnerCore<E> {
                         range_start_key.clone(),
                         range_end_key.clone(),
                     )?;
-                    txn = Self::new_txn();
+                    txn = Self::mvcc_txn_with_dummy_cm();
                 }
             }
 
diff --git a/src/server/raftkv/mod.rs b/src/server/raftkv/mod.rs
index 1e2aeaab033..4beaa2d48d2 100644
--- a/src/server/raftkv/mod.rs
+++ b/src/server/raftkv/mod.rs
@@ -895,7 +895,7 @@ mod tests {
     // index.
     #[test]
     fn test_replica_read_lock_checker_for_single_uuid() {
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let checker = ReplicaReadLockChecker::new(cm);
         let mut m = eraftpb::Message::default();
         m.set_msg_type(MessageType::MsgReadIndex);
@@ -910,7 +910,7 @@ mod tests {
 
     #[test]
     fn test_replica_read_lock_check_when_not_leader() {
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let checker = ReplicaReadLockChecker::new(cm);
         let mut m = eraftpb::Message::default();
         m.set_msg_type(MessageType::MsgReadIndex);
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
index 77ec505c060..55b42df5b71 100644
--- a/src/storage/mod.rs
+++ b/src/storage/mod.rs
@@ -3666,7 +3666,7 @@ impl<E: Engine, L: LockManager, F: KvFormat> TestStorageBuilder<E, L, F> {
             &self.config,
             ReadPool::from(read_pool).handle(),
             self.lock_mgr,
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             DynamicConfigs {
                 pipelined_pessimistic_lock: self.pipelined_pessimistic_lock,
                 in_memory_pessimistic_lock: self.in_memory_pessimistic_lock,
@@ -3702,7 +3702,7 @@ impl<E: Engine, L: LockManager, F: KvFormat> TestStorageBuilder<E, L, F> {
             &self.config,
             ReadPool::from(read_pool).handle(),
             self.lock_mgr,
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             DynamicConfigs {
                 pipelined_pessimistic_lock: self.pipelined_pessimistic_lock,
                 in_memory_pessimistic_lock: self.in_memory_pessimistic_lock,
@@ -3741,7 +3741,7 @@ impl<E: Engine, L: LockManager, F: KvFormat> TestStorageBuilder<E, L, F> {
             &self.config,
             ReadPool::from(read_pool).handle(),
             self.lock_mgr,
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             DynamicConfigs {
                 pipelined_pessimistic_lock: self.pipelined_pessimistic_lock,
                 in_memory_pessimistic_lock: self.in_memory_pessimistic_lock,
diff --git a/src/storage/mvcc/reader/reader.rs b/src/storage/mvcc/reader/reader.rs
index 5e26f964521..d65a775622e 100644
--- a/src/storage/mvcc/reader/reader.rs
+++ b/src/storage/mvcc/reader/reader.rs
@@ -1066,7 +1066,7 @@ pub mod tests {
         pub fn prewrite(&mut self, m: Mutation, pk: &[u8], start_ts: impl Into<TimeStamp>) {
             let snap = self.snapshot();
             let start_ts = start_ts.into();
-            let cm = ConcurrencyManager::new(start_ts);
+            let cm = ConcurrencyManager::new_for_test(start_ts);
             let mut txn = MvccTxn::new(start_ts, cm);
             let mut reader = SnapshotReader::new(start_ts, snap, true);
 
@@ -1091,7 +1091,7 @@ pub mod tests {
         ) {
             let snap = self.snapshot();
             let start_ts = start_ts.into();
-            let cm = ConcurrencyManager::new(start_ts);
+            let cm = ConcurrencyManager::new_for_test(start_ts);
             let mut txn = MvccTxn::new(start_ts, cm);
             let mut reader = SnapshotReader::new(start_ts, snap, true);
 
@@ -1117,7 +1117,7 @@ pub mod tests {
         ) {
             let snap = self.snapshot();
             let for_update_ts = for_update_ts.into();
-            let cm = ConcurrencyManager::new(for_update_ts);
+            let cm = ConcurrencyManager::new_for_test(for_update_ts);
             let start_ts = start_ts.into();
             let mut txn = MvccTxn::new(start_ts, cm);
             let mut reader = SnapshotReader::new(start_ts, snap, true);
@@ -1148,7 +1148,7 @@ pub mod tests {
         ) {
             let snap = self.snapshot();
             let start_ts = start_ts.into();
-            let cm = ConcurrencyManager::new(start_ts);
+            let cm = ConcurrencyManager::new_for_test(start_ts);
             let mut txn = MvccTxn::new(start_ts, cm);
             let mut reader = SnapshotReader::new(start_ts, snap, true);
             commit(&mut txn, &mut reader, Key::from_raw(pk), commit_ts.into()).unwrap();
@@ -1158,7 +1158,7 @@ pub mod tests {
         pub fn rollback(&mut self, pk: &[u8], start_ts: impl Into<TimeStamp>) {
             let snap = self.snapshot();
             let start_ts = start_ts.into();
-            let cm = ConcurrencyManager::new(start_ts);
+            let cm = ConcurrencyManager::new_for_test(start_ts);
             let mut txn = MvccTxn::new(start_ts, cm);
             let mut reader = SnapshotReader::new(start_ts, snap, true);
             cleanup(
@@ -1173,7 +1173,7 @@ pub mod tests {
         }
 
         pub fn gc(&mut self, pk: &[u8], safe_point: impl Into<TimeStamp> + Copy) {
-            let cm = ConcurrencyManager::new(safe_point.into());
+            let cm = ConcurrencyManager::new_for_test(safe_point.into());
             loop {
                 let snap = self.snapshot();
                 let mut txn = MvccTxn::new(safe_point.into(), cm.clone());
@@ -2480,7 +2480,7 @@ pub mod tests {
         ];
         for (i, case) in cases.into_iter().enumerate() {
             let mut engine = TestEngineBuilder::new().build().unwrap();
-            let cm = ConcurrencyManager::new(42.into());
+            let cm = ConcurrencyManager::new_for_test(42.into());
             let mut txn = MvccTxn::new(TimeStamp::new(10), cm.clone());
             for (write_record, put_ts) in case.written.iter() {
                 txn.put_write(
diff --git a/src/storage/mvcc/txn.rs b/src/storage/mvcc/txn.rs
index 7a9a4b82663..3db6ee60713 100644
--- a/src/storage/mvcc/txn.rs
+++ b/src/storage/mvcc/txn.rs
@@ -806,7 +806,7 @@ pub(crate) mod tests {
         let mut engine = TestEngineBuilder::new().build().unwrap();
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
-        let cm = ConcurrencyManager::new(10.into());
+        let cm = ConcurrencyManager::new_for_test(10.into());
         let mut txn = MvccTxn::new(10.into(), cm.clone());
         let mut reader = SnapshotReader::new(10.into(), snapshot, true);
         let key = Key::from_raw(k);
@@ -854,7 +854,7 @@ pub(crate) mod tests {
         must_commit(&mut engine, key, 5, 10);
 
         let snapshot = engine.snapshot(Default::default()).unwrap();
-        let cm = ConcurrencyManager::new(10.into());
+        let cm = ConcurrencyManager::new_for_test(10.into());
         let mut txn = MvccTxn::new(5.into(), cm.clone());
         let mut reader = SnapshotReader::new(5.into(), snapshot, true);
         prewrite(
@@ -1302,7 +1302,7 @@ pub(crate) mod tests {
         let mut engine = TestEngineBuilder::new().build().unwrap();
         let mut engine_clone = engine.clone();
         let ctx = Context::default();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         let mut do_prewrite = || {
             let snapshot = engine_clone.snapshot(Default::default()).unwrap();
@@ -1359,7 +1359,7 @@ pub(crate) mod tests {
     fn test_async_pessimistic_prewrite_primary() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
         let ctx = Context::default();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         must_acquire_pessimistic_lock(&mut engine, b"key", b"key", 2, 2);
 
@@ -1417,7 +1417,7 @@ pub(crate) mod tests {
     #[test]
     fn test_async_commit_pushed_min_commit_ts() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         // Simulate that min_commit_ts is pushed forward larger than latest_ts
         must_acquire_pessimistic_lock_impl(
diff --git a/src/storage/txn/actions/acquire_pessimistic_lock.rs b/src/storage/txn/actions/acquire_pessimistic_lock.rs
index 6c71bb6843a..0704274a7e9 100644
--- a/src/storage/txn/actions/acquire_pessimistic_lock.rs
+++ b/src/storage/txn/actions/acquire_pessimistic_lock.rs
@@ -467,7 +467,7 @@ pub mod tests {
     ) -> MvccResult<PessimisticLockKeyResult> {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
-        let cm = ConcurrencyManager::new(0.into());
+        let cm = ConcurrencyManager::new_for_test(0.into());
         let start_ts = start_ts.into();
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
@@ -538,7 +538,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let min_commit_ts = min_commit_ts.into();
-        let cm = ConcurrencyManager::new(min_commit_ts);
+        let cm = ConcurrencyManager::new_for_test(min_commit_ts);
         let start_ts = start_ts.into();
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
@@ -720,7 +720,7 @@ pub mod tests {
     ) -> MvccError {
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let min_commit_ts = min_commit_ts.into();
-        let cm = ConcurrencyManager::new(min_commit_ts);
+        let cm = ConcurrencyManager::new_for_test(min_commit_ts);
         let start_ts = start_ts.into();
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
@@ -1325,7 +1325,7 @@ pub mod tests {
             for need_value in &[true, false] {
                 for need_check_existence in &[true, false] {
                     let snapshot = engine.snapshot(Default::default()).unwrap();
-                    let cm = ConcurrencyManager::new(start_ts);
+                    let cm = ConcurrencyManager::new_for_test(start_ts);
                     let mut txn = MvccTxn::new(start_ts, cm);
                     let mut reader = SnapshotReader::new(start_ts, snapshot, true);
                     let need_old_value = true;
@@ -1375,7 +1375,7 @@ pub mod tests {
         // Lock @ start ts 9, for update ts 12, commit ts 13
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let min_commit_ts = TimeStamp::zero();
-        let cm = ConcurrencyManager::new(min_commit_ts);
+        let cm = ConcurrencyManager::new_for_test(min_commit_ts);
         let start_ts = TimeStamp::new(9);
         let for_update_ts = TimeStamp::new(12);
         let need_old_value = true;
@@ -1456,7 +1456,7 @@ pub mod tests {
                     let need_value = *need_value;
                     let t = Box::new(move |snapshot, start_ts| {
                         let key = Key::from_raw(key);
-                        let cm = ConcurrencyManager::new(start_ts);
+                        let cm = ConcurrencyManager::new_for_test(start_ts);
                         let mut txn = MvccTxn::new(start_ts, cm);
                         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
                         let need_old_value = true;
@@ -1509,7 +1509,7 @@ pub mod tests {
         // set.
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let min_commit_ts = TimeStamp::zero();
-        let cm = ConcurrencyManager::new(min_commit_ts);
+        let cm = ConcurrencyManager::new_for_test(min_commit_ts);
         let start_ts = TimeStamp::new(15);
         let for_update_ts = TimeStamp::new(15);
         let need_old_value = true;
diff --git a/src/storage/txn/actions/check_data_constraint.rs b/src/storage/txn/actions/check_data_constraint.rs
index 9be878df2ee..96857534df9 100644
--- a/src/storage/txn/actions/check_data_constraint.rs
+++ b/src/storage/txn/actions/check_data_constraint.rs
@@ -63,7 +63,7 @@ mod tests {
     #[test]
     fn test_check_data_constraint() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
         let mut txn = MvccTxn::new(TimeStamp::new(2), cm);
         txn.put_write(
             Key::from_raw(b"a"),
diff --git a/src/storage/txn/actions/cleanup.rs b/src/storage/txn/actions/cleanup.rs
index d28368aa1bf..3bec26df17c 100644
--- a/src/storage/txn/actions/cleanup.rs
+++ b/src/storage/txn/actions/cleanup.rs
@@ -113,7 +113,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let current_ts = current_ts.into();
-        let cm = ConcurrencyManager::new(current_ts);
+        let cm = ConcurrencyManager::new_for_test(current_ts);
         let start_ts = start_ts.into();
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
@@ -129,7 +129,7 @@ pub mod tests {
     ) -> MvccError {
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let current_ts = current_ts.into();
-        let cm = ConcurrencyManager::new(current_ts);
+        let cm = ConcurrencyManager::new_for_test(current_ts);
         let start_ts = start_ts.into();
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
@@ -156,7 +156,7 @@ pub mod tests {
             must_commit(engine, key, gc_fence.prev(), gc_fence);
         }
 
-        let cm = ConcurrencyManager::new(current_ts);
+        let cm = ConcurrencyManager::new_for_test(current_ts);
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
diff --git a/src/storage/txn/actions/commit.rs b/src/storage/txn/actions/commit.rs
index 1685dde1c88..a4225a043cd 100644
--- a/src/storage/txn/actions/commit.rs
+++ b/src/storage/txn/actions/commit.rs
@@ -179,7 +179,7 @@ pub mod tests {
         };
         let snapshot = engine.snapshot(snap_ctx).unwrap();
         let start_ts = start_ts.into();
-        let cm = ConcurrencyManager::new(start_ts);
+        let cm = ConcurrencyManager::new_for_test(start_ts);
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
         let res = commit(&mut txn, &mut reader, Key::from_raw(key), commit_ts.into()).unwrap();
@@ -195,7 +195,7 @@ pub mod tests {
     ) {
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let start_ts = start_ts.into();
-        let cm = ConcurrencyManager::new(start_ts);
+        let cm = ConcurrencyManager::new_for_test(start_ts);
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
         commit(&mut txn, &mut reader, Key::from_raw(key), commit_ts.into()).unwrap_err();
diff --git a/src/storage/txn/actions/flashback_to_version.rs b/src/storage/txn/actions/flashback_to_version.rs
index dddc7cf0d15..87a982dcd85 100644
--- a/src/storage/txn/actions/flashback_to_version.rs
+++ b/src/storage/txn/actions/flashback_to_version.rs
@@ -326,7 +326,7 @@ pub mod tests {
         let key_locks =
             flashback_to_version_read_lock(&mut reader, key, Some(next_key).as_ref(), start_ts)
                 .unwrap();
-        let cm = ConcurrencyManager::new(TimeStamp::zero());
+        let cm = ConcurrencyManager::new_for_test(TimeStamp::zero());
         let mut txn = MvccTxn::new(start_ts, cm);
         rollback_locks(&mut txn, snapshot, key_locks).unwrap();
         let rows = txn.modifies.len();
@@ -341,7 +341,7 @@ pub mod tests {
         start_ts: impl Into<TimeStamp>,
     ) -> usize {
         let (version, start_ts) = (version.into(), start_ts.into());
-        let cm = ConcurrencyManager::new(TimeStamp::zero());
+        let cm = ConcurrencyManager::new_for_test(TimeStamp::zero());
         let mut txn = MvccTxn::new(start_ts, cm);
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let ctx = Context::default();
@@ -388,7 +388,7 @@ pub mod tests {
             commit_ts,
         )
         .unwrap();
-        let cm = ConcurrencyManager::new(TimeStamp::zero());
+        let cm = ConcurrencyManager::new_for_test(TimeStamp::zero());
         let mut txn = MvccTxn::new(start_ts, cm);
         flashback_to_version_write(&mut txn, &mut reader, keys, version, start_ts, commit_ts)
             .unwrap();
@@ -405,7 +405,7 @@ pub mod tests {
         commit_ts: impl Into<TimeStamp>,
     ) -> usize {
         let (version, start_ts, commit_ts) = (version.into(), start_ts.into(), commit_ts.into());
-        let cm = ConcurrencyManager::new(TimeStamp::zero());
+        let cm = ConcurrencyManager::new_for_test(TimeStamp::zero());
         let mut txn = MvccTxn::new(start_ts, cm);
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let ctx = Context::default();
diff --git a/src/storage/txn/actions/gc.rs b/src/storage/txn/actions/gc.rs
index 8c24baf7d5b..890d5ed7b46 100644
--- a/src/storage/txn/actions/gc.rs
+++ b/src/storage/txn/actions/gc.rs
@@ -140,7 +140,7 @@ pub mod tests {
     pub fn must_succeed<E: Engine>(engine: &mut E, key: &[u8], safe_point: impl Into<TimeStamp>) {
         let ctx = SnapContext::default();
         let snapshot = engine.snapshot(ctx).unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut txn = MvccTxn::new(TimeStamp::zero(), cm);
         let mut reader = MvccReader::new(snapshot, Some(ScanMode::Forward), true);
         gc(&mut txn, &mut reader, Key::from_raw(key), safe_point.into()).unwrap();
diff --git a/src/storage/txn/actions/prewrite.rs b/src/storage/txn/actions/prewrite.rs
index b618f63892b..117f86dd2a6 100644
--- a/src/storage/txn/actions/prewrite.rs
+++ b/src/storage/txn/actions/prewrite.rs
@@ -1000,7 +1000,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let ts = ts.into();
-        let cm = ConcurrencyManager::new(ts);
+        let cm = ConcurrencyManager::new_for_test(ts);
         let mut txn = MvccTxn::new(ts, cm);
         let mut reader = SnapshotReader::new(ts, snapshot, true);
 
@@ -1034,7 +1034,7 @@ pub mod tests {
     ) -> Result<()> {
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let ts = ts.into();
-        let cm = ConcurrencyManager::new(ts);
+        let cm = ConcurrencyManager::new_for_test(ts);
         let mut txn = MvccTxn::new(ts, cm);
         let mut reader = SnapshotReader::new(ts, snapshot, true);
 
@@ -1054,7 +1054,7 @@ pub mod tests {
     #[test]
     fn test_async_commit_prewrite_check_max_commit_ts() {
         let mut engine = crate::storage::TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let mut txn = MvccTxn::new(10.into(), cm.clone());
@@ -1101,7 +1101,7 @@ pub mod tests {
     #[test]
     fn test_async_commit_prewrite_min_commit_ts() {
         let mut engine = crate::storage::TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(41.into());
+        let cm = ConcurrencyManager::new_for_test(41.into());
         let snapshot = engine.snapshot(Default::default()).unwrap();
 
         // should_not_write mutations don't write locks or change data so that they
@@ -1239,7 +1239,7 @@ pub mod tests {
     #[test]
     fn test_1pc_check_max_commit_ts() {
         let mut engine = crate::storage::TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         let snapshot = engine.snapshot(Default::default()).unwrap();
 
@@ -1294,7 +1294,7 @@ pub mod tests {
     ) -> Result<()> {
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let ts = ts.into();
-        let cm = ConcurrencyManager::new(ts);
+        let cm = ConcurrencyManager::new_for_test(ts);
         let mut txn = MvccTxn::new(ts, cm);
         let mut reader = SnapshotReader::new(ts, snapshot, false);
 
@@ -1326,7 +1326,7 @@ pub mod tests {
     #[test]
     fn test_async_commit_pessimistic_prewrite_check_max_commit_ts() {
         let mut engine = crate::storage::TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         must_acquire_pessimistic_lock(&mut engine, b"k1", b"k1", 10, 10);
         must_acquire_pessimistic_lock(&mut engine, b"k2", b"k1", 10, 10);
@@ -1379,7 +1379,7 @@ pub mod tests {
     #[test]
     fn test_1pc_pessimistic_prewrite_check_max_commit_ts() {
         let mut engine = crate::storage::TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         must_acquire_pessimistic_lock(&mut engine, b"k1", b"k1", 10, 10);
         must_acquire_pessimistic_lock(&mut engine, b"k2", b"k1", 10, 10);
@@ -1432,7 +1432,7 @@ pub mod tests {
     #[test]
     fn test_prewrite_check_gc_fence() {
         let mut engine = crate::storage::TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
 
         // PUT,           Read
         //  `------^
@@ -1850,7 +1850,7 @@ pub mod tests {
                 txn_source: 0,
             };
             let snapshot = engine.snapshot(Default::default()).unwrap();
-            let cm = ConcurrencyManager::new(start_ts);
+            let cm = ConcurrencyManager::new_for_test(start_ts);
             let mut txn = MvccTxn::new(start_ts, cm);
             let mut reader = SnapshotReader::new(start_ts, snapshot, true);
             let (_, old_value) = prewrite(
@@ -1906,7 +1906,7 @@ pub mod tests {
             txn_source: 0,
         };
         let snapshot = engine.snapshot(Default::default()).unwrap();
-        let cm = ConcurrencyManager::new(start_ts);
+        let cm = ConcurrencyManager::new_for_test(start_ts);
         let mut txn = MvccTxn::new(start_ts, cm);
         let mut reader = SnapshotReader::new(start_ts, snapshot, true);
         let (_, old_value) = prewrite(
@@ -2031,7 +2031,7 @@ pub mod tests {
             key,
             require_old_value_none,
             vec![Box::new(move |snapshot, start_ts| {
-                let cm = ConcurrencyManager::new(start_ts);
+                let cm = ConcurrencyManager::new_for_test(start_ts);
                 let mut txn = MvccTxn::new(start_ts, cm);
                 let mut reader = SnapshotReader::new(start_ts, snapshot, true);
                 let txn_props = TransactionProperties {
@@ -2069,7 +2069,7 @@ pub mod tests {
             key,
             require_old_value_none,
             vec![Box::new(move |snapshot, start_ts| {
-                let cm = ConcurrencyManager::new(start_ts);
+                let cm = ConcurrencyManager::new_for_test(start_ts);
                 let mut txn = MvccTxn::new(start_ts, cm);
                 let mut reader = SnapshotReader::new(start_ts, snapshot, true);
                 let txn_props = TransactionProperties {
@@ -2770,7 +2770,7 @@ pub mod tests {
     #[test]
     fn test_1pc_set_lock_use_one_pc() {
         let mut engine = crate::storage::TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
 
         let snapshot = engine.snapshot(Default::default()).unwrap();
 
diff --git a/src/storage/txn/commands/acquire_pessimistic_lock_resumed.rs b/src/storage/txn/commands/acquire_pessimistic_lock_resumed.rs
index 336ed6d43f5..ce937d7dceb 100644
--- a/src/storage/txn/commands/acquire_pessimistic_lock_resumed.rs
+++ b/src/storage/txn/commands/acquire_pessimistic_lock_resumed.rs
@@ -261,7 +261,7 @@ mod tests {
     ) -> PessimisticLockResults {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
-        let cm = ConcurrencyManager::new(TimeStamp::zero());
+        let cm = ConcurrencyManager::new_for_test(TimeStamp::zero());
 
         let items_info: Vec<_> = lock_wait_entries
             .iter()
diff --git a/src/storage/txn/commands/atomic_store.rs b/src/storage/txn/commands/atomic_store.rs
index 7626ee2a6d7..8262b13a249 100644
--- a/src/storage/txn/commands/atomic_store.rs
+++ b/src/storage/txn/commands/atomic_store.rs
@@ -93,7 +93,7 @@ mod tests {
 
     fn test_atomic_process_write_impl<F: KvFormat>() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
         let raw_keys = [b"ra", b"rz"];
         let raw_values = [b"valuea", b"valuez"];
         let ts_provider = super::super::test_util::gen_ts_provider(F::TAG);
diff --git a/src/storage/txn/commands/check_secondary_locks.rs b/src/storage/txn/commands/check_secondary_locks.rs
index 6664f16bdc6..1cb5f18c2da 100644
--- a/src/storage/txn/commands/check_secondary_locks.rs
+++ b/src/storage/txn/commands/check_secondary_locks.rs
@@ -267,7 +267,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let lock_ts = lock_ts.into();
-        let cm = ConcurrencyManager::new(lock_ts);
+        let cm = ConcurrencyManager::new_for_test(lock_ts);
         let command = crate::storage::txn::commands::CheckSecondaryLocks {
             ctx: ctx.clone(),
             keys: vec![Key::from_raw(key)],
@@ -301,7 +301,7 @@ pub mod tests {
         let mut engine = TestEngineBuilder::new().build().unwrap();
         let mut engine_clone = engine.clone();
         let ctx = Context::default();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
 
         let mut check_secondary = |key, ts| {
             let snapshot = engine_clone.snapshot(Default::default()).unwrap();
diff --git a/src/storage/txn/commands/check_txn_status.rs b/src/storage/txn/commands/check_txn_status.rs
index fbb1b7187f1..a41feff7cd4 100644
--- a/src/storage/txn/commands/check_txn_status.rs
+++ b/src/storage/txn/commands/check_txn_status.rs
@@ -209,7 +209,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let current_ts = current_ts.into();
-        let cm = ConcurrencyManager::new(current_ts);
+        let cm = ConcurrencyManager::new_for_test(current_ts);
         let lock_ts: TimeStamp = lock_ts.into();
         let command = crate::storage::txn::commands::CheckTxnStatus {
             ctx: Context::default(),
@@ -263,7 +263,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let current_ts = current_ts.into();
-        let cm = ConcurrencyManager::new(current_ts);
+        let cm = ConcurrencyManager::new_for_test(current_ts);
         let lock_ts: TimeStamp = lock_ts.into();
         let command = crate::storage::txn::commands::CheckTxnStatus {
             ctx,
diff --git a/src/storage/txn/commands/compare_and_swap.rs b/src/storage/txn/commands/compare_and_swap.rs
index 89b3287219a..72d0077d5a7 100644
--- a/src/storage/txn/commands/compare_and_swap.rs
+++ b/src/storage/txn/commands/compare_and_swap.rs
@@ -157,7 +157,7 @@ mod tests {
     fn test_cas_basic_impl<F: KvFormat>() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
         let ts_provider = super::super::test_util::gen_ts_provider(F::TAG);
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
         let key = b"rk";
 
         let encoded_key = F::encode_raw_key(key, None);
@@ -249,7 +249,7 @@ mod tests {
         let mut engine = TestEngineBuilder::new().build().unwrap();
         let ts_provider = super::super::test_util::gen_ts_provider(F::TAG);
 
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
         let raw_key = b"rk";
         let raw_value = b"valuek";
         let ttl = 30;
diff --git a/src/storage/txn/commands/mod.rs b/src/storage/txn/commands/mod.rs
index 48654c826a7..be7bb8f1bde 100644
--- a/src/storage/txn/commands/mod.rs
+++ b/src/storage/txn/commands/mod.rs
@@ -961,7 +961,7 @@ pub mod test_util {
         start_ts: u64,
         one_pc_max_commit_ts: Option<u64>,
     ) -> Result<PrewriteResult> {
-        let cm = ConcurrencyManager::new(start_ts.into());
+        let cm = ConcurrencyManager::new_for_test(start_ts.into());
         prewrite_with_cm(
             engine,
             cm,
@@ -1004,7 +1004,7 @@ pub mod test_util {
         for_update_ts: u64,
         one_pc_max_commit_ts: Option<u64>,
     ) -> Result<PrewriteResult> {
-        let cm = ConcurrencyManager::new(start_ts.into());
+        let cm = ConcurrencyManager::new_for_test(start_ts.into());
         pessimistic_prewrite_with_cm(
             engine,
             cm,
@@ -1064,7 +1064,7 @@ pub mod test_util {
                 .into_iter()
                 .map(|(size, ts)| (size, TimeStamp::from(ts))),
         );
-        let cm = ConcurrencyManager::new(start_ts.into());
+        let cm = ConcurrencyManager::new_for_test(start_ts.into());
         prewrite_command(engine, cm, statistics, cmd)
     }
 
@@ -1077,7 +1077,7 @@ pub mod test_util {
     ) -> Result<()> {
         let ctx = Context::default();
         let snap = engine.snapshot(Default::default())?;
-        let concurrency_manager = ConcurrencyManager::new(lock_ts.into());
+        let concurrency_manager = ConcurrencyManager::new_for_test(lock_ts.into());
         let cmd = Commit::new(
             keys,
             TimeStamp::from(lock_ts),
@@ -1109,7 +1109,7 @@ pub mod test_util {
     ) -> Result<()> {
         let ctx = Context::default();
         let snap = engine.snapshot(Default::default())?;
-        let concurrency_manager = ConcurrencyManager::new(start_ts.into());
+        let concurrency_manager = ConcurrencyManager::new_for_test(start_ts.into());
         let cmd = Rollback::new(keys, TimeStamp::from(start_ts), ctx);
         let context = WriteContext {
             lock_mgr: &MockLockManager::new(),
diff --git a/src/storage/txn/commands/pessimistic_rollback.rs b/src/storage/txn/commands/pessimistic_rollback.rs
index f51319190cb..77e9beeb06a 100644
--- a/src/storage/txn/commands/pessimistic_rollback.rs
+++ b/src/storage/txn/commands/pessimistic_rollback.rs
@@ -157,7 +157,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let for_update_ts = for_update_ts.into();
-        let cm = ConcurrencyManager::new(for_update_ts);
+        let cm = ConcurrencyManager::new_for_test(for_update_ts);
         let start_ts = start_ts.into();
         let command = crate::storage::txn::commands::PessimisticRollback {
             ctx: ctx.clone(),
diff --git a/src/storage/txn/commands/prewrite.rs b/src/storage/txn/commands/prewrite.rs
index d9c7a65921b..83b27af9c24 100644
--- a/src/storage/txn/commands/prewrite.rs
+++ b/src/storage/txn/commands/prewrite.rs
@@ -1208,7 +1208,7 @@ mod tests {
         use crate::storage::mvcc::tests::{must_get, must_get_commit_ts, must_unlocked};
 
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
 
         let key = b"k";
         let value = b"v";
@@ -1244,7 +1244,7 @@ mod tests {
         use crate::storage::mvcc::tests::{must_get, must_get_commit_ts, must_unlocked};
 
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
 
         let key = b"k";
         let value = b"v";
@@ -1345,7 +1345,7 @@ mod tests {
     #[test]
     fn test_prewrite_pessimsitic_1pc() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
         let key = b"k";
         let value = b"v";
 
@@ -1480,7 +1480,7 @@ mod tests {
     #[test]
     fn test_prewrite_async_commit() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
 
         let key = b"k";
         let value = b"v";
@@ -1546,7 +1546,7 @@ mod tests {
     #[test]
     fn test_prewrite_pessimsitic_async_commit() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
 
         let key = b"k";
         let value = b"v";
@@ -1668,7 +1668,7 @@ mod tests {
             () => {
                 WriteContext {
                     lock_mgr: &MockLockManager::new(),
-                    concurrency_manager: ConcurrencyManager::new(10.into()),
+                    concurrency_manager: ConcurrencyManager::new_for_test(10.into()),
                     extra_op: ExtraOp::Noop,
                     statistics: &mut Statistics::default(),
                     async_apply_prewrite: false,
@@ -1717,7 +1717,7 @@ mod tests {
     // this test shows which stage in raft can we return the response
     #[test]
     fn test_response_stage() {
-        let cm = ConcurrencyManager::new(42.into());
+        let cm = ConcurrencyManager::new_for_test(42.into());
         let start_ts = TimeStamp::new(10);
         let keys = [b"k1", b"k2"];
         let values = [b"v1", b"v2"];
@@ -1859,7 +1859,7 @@ mod tests {
     fn test_prewrite_should_not_exist() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
         // concurency_manager.max_tx = 5
-        let cm = ConcurrencyManager::new(5.into());
+        let cm = ConcurrencyManager::new_for_test(5.into());
         let mut statistics = Statistics::default();
 
         let (key, value) = (b"k", b"val");
@@ -1911,7 +1911,7 @@ mod tests {
     #[test]
     fn test_optimistic_prewrite_committed_transaction() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut statistics = Statistics::default();
 
         let key = b"k";
@@ -2004,7 +2004,7 @@ mod tests {
     #[test]
     fn test_pessimistic_prewrite_committed_transaction() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut statistics = Statistics::default();
 
         let key = b"k";
@@ -2121,7 +2121,7 @@ mod tests {
     #[test]
     fn test_repeated_pessimistic_prewrite_1pc() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut statistics = Statistics::default();
 
         must_acquire_pessimistic_lock(&mut engine, b"k2", b"k2", 5, 5);
@@ -2170,7 +2170,7 @@ mod tests {
     #[test]
     fn test_repeated_prewrite_non_pessimistic_lock() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut statistics = Statistics::default();
 
         let cm = &cm;
@@ -2340,7 +2340,7 @@ mod tests {
     #[test]
     fn test_prewrite_rolledback_transaction() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut statistics = Statistics::default();
 
         let k1 = b"k1";
@@ -2478,7 +2478,7 @@ mod tests {
 
         // Txn1 continues. If the two keys are sent in the single prewrite request, the
         // AssertionFailed error won't be returned since there are other error.
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut stat = Statistics::default();
         // Two keys in single request:
         let cmd = PrewritePessimistic::with_defaults(
@@ -2600,7 +2600,7 @@ mod tests {
         );
         let context = WriteContext {
             lock_mgr: &MockLockManager::new(),
-            concurrency_manager: ConcurrencyManager::new(20.into()),
+            concurrency_manager: ConcurrencyManager::new_for_test(20.into()),
             extra_op: ExtraOp::Noop,
             statistics: &mut statistics,
             async_apply_prewrite: false,
@@ -2621,7 +2621,7 @@ mod tests {
     #[test]
     fn test_repeated_prewrite_commit_ts_too_large() {
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         let mut statistics = Statistics::default();
 
         // First, prewrite and commit normally.
@@ -2691,7 +2691,7 @@ mod tests {
         use crate::storage::txn::sched_pool::set_tls_feature_gate;
 
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
 
         let key = b"k";
         let value = b"v";
@@ -2773,7 +2773,7 @@ mod tests {
         use crate::storage::txn::sched_pool::set_tls_feature_gate;
 
         let mut engine = TestEngineBuilder::new().build().unwrap();
-        let cm = concurrency_manager::ConcurrencyManager::new(1.into());
+        let cm = concurrency_manager::ConcurrencyManager::new_for_test(1.into());
 
         let key = b"k";
         let value = b"v";
diff --git a/src/storage/txn/commands/txn_heart_beat.rs b/src/storage/txn/commands/txn_heart_beat.rs
index 11c53a05878..a41492d7020 100644
--- a/src/storage/txn/commands/txn_heart_beat.rs
+++ b/src/storage/txn/commands/txn_heart_beat.rs
@@ -157,7 +157,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let start_ts = start_ts.into();
-        let cm = ConcurrencyManager::new(start_ts);
+        let cm = ConcurrencyManager::new_for_test(start_ts);
         let command = TxnHeartBeat {
             ctx: Context::default(),
             primary_key: Key::from_raw(primary_key),
@@ -200,7 +200,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let start_ts = start_ts.into();
-        let cm = ConcurrencyManager::new(start_ts);
+        let cm = ConcurrencyManager::new_for_test(start_ts);
         let command = TxnHeartBeat {
             ctx,
             primary_key: Key::from_raw(primary_key),
@@ -289,7 +289,7 @@ pub mod tests {
         let ctx = Context::default();
         let snapshot = engine.snapshot(Default::default()).unwrap();
         let start_ts = 5.into();
-        let cm = ConcurrencyManager::new(start_ts);
+        let cm = ConcurrencyManager::new_for_test(start_ts);
         let command = TxnHeartBeat {
             ctx: ctx.clone(),
             primary_key: Key::from_raw(k),
diff --git a/src/storage/txn/scheduler.rs b/src/storage/txn/scheduler.rs
index 955e8395965..eee8370d87b 100644
--- a/src/storage/txn/scheduler.rs
+++ b/src/storage/txn/scheduler.rs
@@ -2222,7 +2222,7 @@ mod tests {
             TxnScheduler::new(
                 engine.clone(),
                 MockLockManager::new(),
-                ConcurrencyManager::new(1.into()),
+                ConcurrencyManager::new_for_test(1.into()),
                 &config,
                 DynamicConfigs {
                     pipelined_pessimistic_lock: Arc::new(AtomicBool::new(true)),
@@ -2575,7 +2575,7 @@ mod tests {
         let scheduler = TxnScheduler::new(
             engine,
             MockLockManager::new(),
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             &config,
             DynamicConfigs {
                 pipelined_pessimistic_lock: Arc::new(AtomicBool::new(false)),
diff --git a/src/storage/txn/store.rs b/src/storage/txn/store.rs
index 99c9c7d887c..ed7e9105076 100644
--- a/src/storage/txn/store.rs
+++ b/src/storage/txn/store.rs
@@ -706,7 +706,7 @@ mod tests {
 
             // do prewrite.
             {
-                let cm = ConcurrencyManager::new(START_TS);
+                let cm = ConcurrencyManager::new_for_test(START_TS);
                 let mut txn = MvccTxn::new(START_TS, cm);
                 let mut reader = SnapshotReader::new(START_TS, self.snapshot.clone(), true);
                 for key in &self.keys {
@@ -740,7 +740,7 @@ mod tests {
             self.refresh_snapshot();
             // do commit
             {
-                let cm = ConcurrencyManager::new(START_TS);
+                let cm = ConcurrencyManager::new_for_test(START_TS);
                 let mut txn = MvccTxn::new(START_TS, cm);
                 let mut reader = SnapshotReader::new(START_TS, self.snapshot.clone(), true);
                 for key in &self.keys {


Test patch:
diff --git a/components/concurrency_manager/tests/memory_usage.rs b/components/concurrency_manager/tests/memory_usage.rs
index 76fab7e185c..6822c2fdc52 100644
--- a/components/concurrency_manager/tests/memory_usage.rs
+++ b/components/concurrency_manager/tests/memory_usage.rs
@@ -28,7 +28,7 @@ fn test_memory_usage() {
     const LOCK_COUNT: usize = 100 * 1024 * 1024 / KEY_LEN;
 
     // Let system collect the memory to avoid drop time at the end of the test.
-    let cm = ManuallyDrop::new(ConcurrencyManager::new(1.into()));
+    let cm = ManuallyDrop::new(ConcurrencyManager::new_for_test(1.into()));
 
     const THR_NUM: usize = 8;
     let mut ths = Vec::with_capacity(THR_NUM);
diff --git a/components/raftstore-v2/tests/integrations/cluster.rs b/components/raftstore-v2/tests/integrations/cluster.rs
index 1a52e86f098..0cc14abf2c2 100644
--- a/components/raftstore-v2/tests/integrations/cluster.rs
+++ b/components/raftstore-v2/tests/integrations/cluster.rs
@@ -422,7 +422,7 @@ impl TestNode {
             cfg,
             cop_cfg,
             trans,
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             None,
             &self.logger,
             resource_ctl,
diff --git a/components/test_coprocessor/src/fixture.rs b/components/test_coprocessor/src/fixture.rs
index 08299db2803..863e24b980f 100644
--- a/components/test_coprocessor/src/fixture.rs
+++ b/components/test_coprocessor/src/fixture.rs
@@ -189,7 +189,7 @@ fn init_data_with_details_impl<E: Engine>(
         &CoprReadPoolConfig::default_for_test(),
         store.get_engine(),
     ));
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     let limiter = Arc::new(QuotaLimiter::default());
     let copr = Endpoint::new(
         cfg,
diff --git a/components/test_raftstore-v2/src/node.rs b/components/test_raftstore-v2/src/node.rs
index 9e2d30aba02..2bb96622066 100644
--- a/components/test_raftstore-v2/src/node.rs
+++ b/components/test_raftstore-v2/src/node.rs
@@ -272,7 +272,7 @@ impl<EK: KvEngine> Simulator<EK> for NodeCluster<EK> {
         //     f(node_id, &mut coprocessor_host);
         // }
 
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         self.concurrency_managers.insert(node_id, cm.clone());
 
         ReplicaReadLockChecker::new(cm.clone()).register(&mut coprocessor_host);
diff --git a/components/test_raftstore-v2/src/server.rs b/components/test_raftstore-v2/src/server.rs
index 4c2906a9be9..8be0cc57dc4 100644
--- a/components/test_raftstore-v2/src/server.rs
+++ b/components/test_raftstore-v2/src/server.rs
@@ -461,7 +461,7 @@ impl<EK: KvEngine> ServerCluster<EK> {
 
         let latest_ts =
             block_on(self.pd_client.get_tso()).expect("failed to get timestamp from PD");
-        let concurrency_manager = ConcurrencyManager::new(latest_ts);
+        let concurrency_manager = ConcurrencyManager::new_for_test(latest_ts);
 
         let (tx, _rx) = std::sync::mpsc::channel();
         let mut gc_worker = GcWorker::new(
diff --git a/components/test_raftstore/src/node.rs b/components/test_raftstore/src/node.rs
index f512a1be228..68fd05134f6 100644
--- a/components/test_raftstore/src/node.rs
+++ b/components/test_raftstore/src/node.rs
@@ -305,7 +305,7 @@ impl Simulator for NodeCluster {
             f(node_id, &mut coprocessor_host);
         }
 
-        let cm = ConcurrencyManager::new(1.into());
+        let cm = ConcurrencyManager::new_for_test(1.into());
         self.concurrency_managers.insert(node_id, cm.clone());
         ReplicaReadLockChecker::new(cm.clone()).register(&mut coprocessor_host);
 
diff --git a/components/test_raftstore/src/server.rs b/components/test_raftstore/src/server.rs
index aa852b91fa4..3e8220271f4 100644
--- a/components/test_raftstore/src/server.rs
+++ b/components/test_raftstore/src/server.rs
@@ -394,7 +394,7 @@ impl ServerCluster {
 
         let latest_ts =
             block_on(self.pd_client.get_tso()).expect("failed to get timestamp from PD");
-        let concurrency_manager = ConcurrencyManager::new(latest_ts);
+        let concurrency_manager = ConcurrencyManager::new_for_test(latest_ts);
 
         let (tx, rx) = std::sync::mpsc::channel();
         let mut gc_worker = GcWorker::new(
diff --git a/src/storage/txn/actions/tests.rs b/src/storage/txn/actions/tests.rs
index 3bb2bbf2c2a..ffc8b3b1692 100644
--- a/src/storage/txn/actions/tests.rs
+++ b/src/storage/txn/actions/tests.rs
@@ -136,7 +136,7 @@ pub fn must_prewrite_put_impl_with_should_not_exist<E: Engine>(
         ..Default::default()
     };
     let snapshot = engine.snapshot(snap_ctx).unwrap();
-    let cm = ConcurrencyManager::new(ts);
+    let cm = ConcurrencyManager::new_for_test(ts);
     let mut txn = MvccTxn::new(ts, cm);
     let mut reader = SnapshotReader::new(ts, snapshot, true);
 
@@ -256,7 +256,7 @@ pub fn flush_put_impl_with_assertion<E: Engine>(
         Context::new(),
     );
     let mut statistics = Statistics::default();
-    let cm = ConcurrencyManager::new(start_ts);
+    let cm = ConcurrencyManager::new_for_test(start_ts);
     let context = WriteContext {
         lock_mgr: &MockLockManager::new(),
         concurrency_manager: cm.clone(),
@@ -675,7 +675,7 @@ pub fn must_prewrite_put_err_impl_with_should_not_exist<E: Engine>(
 ) -> Error {
     let snapshot = engine.snapshot(Default::default()).unwrap();
     let for_update_ts = for_update_ts.into();
-    let cm = ConcurrencyManager::new(for_update_ts);
+    let cm = ConcurrencyManager::new_for_test(for_update_ts);
     let ts = ts.into();
     let mut txn = MvccTxn::new(ts, cm);
     let mut reader = SnapshotReader::new(ts, snapshot, true);
@@ -854,7 +854,7 @@ fn must_prewrite_delete_impl<E: Engine>(
     };
     let snapshot = engine.snapshot(snap_ctx).unwrap();
     let for_update_ts = for_update_ts.into();
-    let cm = ConcurrencyManager::new(for_update_ts);
+    let cm = ConcurrencyManager::new_for_test(for_update_ts);
     let ts = ts.into();
     let mut txn = MvccTxn::new(ts, cm);
     let mut reader = SnapshotReader::new(ts, snapshot, true);
@@ -933,7 +933,7 @@ fn must_prewrite_lock_impl<E: Engine>(
     let ctx = Context::default();
     let snapshot = engine.snapshot(Default::default()).unwrap();
     let for_update_ts = for_update_ts.into();
-    let cm = ConcurrencyManager::new(for_update_ts);
+    let cm = ConcurrencyManager::new_for_test(for_update_ts);
     let ts = ts.into();
     let mut txn = MvccTxn::new(ts, cm);
     let mut reader = SnapshotReader::new(ts, snapshot, true);
@@ -972,7 +972,7 @@ pub fn must_prewrite_lock_err<E: Engine>(
 ) {
     let snapshot = engine.snapshot(Default::default()).unwrap();
     let ts = ts.into();
-    let cm = ConcurrencyManager::new(ts);
+    let cm = ConcurrencyManager::new_for_test(ts);
     let mut txn = MvccTxn::new(ts, cm);
     let mut reader = SnapshotReader::new(ts, snapshot, true);
 
@@ -1008,7 +1008,7 @@ pub fn must_rollback<E: Engine>(
     let ctx = Context::default();
     let snapshot = engine.snapshot(Default::default()).unwrap();
     let start_ts = start_ts.into();
-    let cm = ConcurrencyManager::new(start_ts);
+    let cm = ConcurrencyManager::new_for_test(start_ts);
     let mut txn = MvccTxn::new(start_ts, cm);
     let mut reader = SnapshotReader::new(start_ts, snapshot, true);
     txn::cleanup(
@@ -1025,7 +1025,7 @@ pub fn must_rollback<E: Engine>(
 pub fn must_rollback_err<E: Engine>(engine: &mut E, key: &[u8], start_ts: impl Into<TimeStamp>) {
     let snapshot = engine.snapshot(Default::default()).unwrap();
     let start_ts = start_ts.into();
-    let cm = ConcurrencyManager::new(start_ts);
+    let cm = ConcurrencyManager::new_for_test(start_ts);
     let mut txn = MvccTxn::new(start_ts, cm);
     let mut reader = SnapshotReader::new(start_ts, snapshot, true);
     txn::cleanup(
diff --git a/tests/benches/hierarchy/mvcc/mod.rs b/tests/benches/hierarchy/mvcc/mod.rs
index 92dacfe6dc9..2e2c79854a0 100644
--- a/tests/benches/hierarchy/mvcc/mod.rs
+++ b/tests/benches/hierarchy/mvcc/mod.rs
@@ -25,7 +25,7 @@ where
     let ctx = Context::default();
     let snapshot = engine.snapshot(Default::default()).unwrap();
     let start_ts = start_ts.into();
-    let cm = ConcurrencyManager::new(start_ts);
+    let cm = ConcurrencyManager::new_for_test(start_ts);
     let mut txn = MvccTxn::new(start_ts, cm);
     let mut reader = SnapshotReader::new(start_ts, snapshot, true);
 
@@ -69,7 +69,7 @@ where
 
 fn mvcc_prewrite<E: Engine, F: EngineFactory<E>>(b: &mut Bencher<'_>, config: &BenchConfig<F>) {
     let mut engine = config.engine_factory.build();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || {
             let mutations: Vec<(Mutation, Vec<u8>)> = KvGenerator::with_seed(
@@ -119,7 +119,7 @@ fn mvcc_prewrite<E: Engine, F: EngineFactory<E>>(b: &mut Bencher<'_>, config: &B
 
 fn mvcc_commit<E: Engine, F: EngineFactory<E>>(b: &mut Bencher<'_>, config: &BenchConfig<F>) {
     let mut engine = config.engine_factory.build();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || setup_prewrite(&mut engine, config, 1),
         |(snapshot, keys)| {
@@ -138,7 +138,7 @@ fn mvcc_rollback_prewrote<E: Engine, F: EngineFactory<E>>(
     config: &BenchConfig<F>,
 ) {
     let mut engine = config.engine_factory.build();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || setup_prewrite(&mut engine, config, 1),
         |(snapshot, keys)| {
@@ -164,7 +164,7 @@ fn mvcc_rollback_conflict<E: Engine, F: EngineFactory<E>>(
     config: &BenchConfig<F>,
 ) {
     let mut engine = config.engine_factory.build();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || setup_prewrite(&mut engine, config, 2),
         |(snapshot, keys)| {
@@ -190,7 +190,7 @@ fn mvcc_rollback_non_prewrote<E: Engine, F: EngineFactory<E>>(
     config: &BenchConfig<F>,
 ) {
     let mut engine = config.engine_factory.build();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || {
             let kvs = KvGenerator::with_seed(
diff --git a/tests/benches/hierarchy/txn/mod.rs b/tests/benches/hierarchy/txn/mod.rs
index 1a4d047562d..2f86361c419 100644
--- a/tests/benches/hierarchy/txn/mod.rs
+++ b/tests/benches/hierarchy/txn/mod.rs
@@ -26,7 +26,7 @@ where
 
     let snapshot = engine.snapshot(Default::default()).unwrap();
     let start_ts = start_ts.into();
-    let cm = ConcurrencyManager::new(start_ts);
+    let cm = ConcurrencyManager::new_for_test(start_ts);
     let mut txn = MvccTxn::new(start_ts, cm);
     let mut reader = SnapshotReader::new(start_ts, snapshot, true);
 
@@ -65,7 +65,7 @@ where
 fn txn_prewrite<E: Engine, F: EngineFactory<E>>(b: &mut Bencher<'_>, config: &BenchConfig<F>) {
     let mut engine = config.engine_factory.build();
     let ctx = Context::default();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || {
             let mutations: Vec<(Mutation, Vec<u8>)> =
@@ -116,7 +116,7 @@ fn txn_commit<E: Engine, F: EngineFactory<E>>(b: &mut Bencher<'_>, config: &Benc
     let mut engine = config.engine_factory.build();
     let mut engine_clone = engine.clone();
     let ctx = Context::default();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || setup_prewrite(&mut engine_clone, config, 1),
         |keys| {
@@ -140,7 +140,7 @@ fn txn_rollback_prewrote<E: Engine, F: EngineFactory<E>>(
     let mut engine = config.engine_factory.build();
     let mut engine_clone = engine.clone();
     let ctx = Context::default();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || setup_prewrite(&mut engine_clone, config, 1),
         |keys| {
@@ -164,7 +164,7 @@ fn txn_rollback_conflict<E: Engine, F: EngineFactory<E>>(
     let mut engine = config.engine_factory.build();
     let mut engine_clone = engine.clone();
     let ctx = Context::default();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || setup_prewrite(&mut engine_clone, config, 2),
         |keys| {
@@ -187,7 +187,7 @@ fn txn_rollback_non_prewrote<E: Engine, F: EngineFactory<E>>(
 ) {
     let mut engine = config.engine_factory.build();
     let ctx = Context::default();
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     b.iter_batched(
         || {
             let kvs = KvGenerator::new(config.key_length, config.value_length)
diff --git a/tests/integrations/config/dynamic/raftstore.rs b/tests/integrations/config/dynamic/raftstore.rs
index 7c39487d4dd..49e44153119 100644
--- a/tests/integrations/config/dynamic/raftstore.rs
+++ b/tests/integrations/config/dynamic/raftstore.rs
@@ -110,7 +110,7 @@ fn start_raftstore(
             Worker::new("split"),
             AutoSplitController::default(),
             Arc::default(),
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             CollectorRegHandle::new_for_test(),
             HealthController::new(),
             None,
diff --git a/tests/integrations/raftstore/test_bootstrap.rs b/tests/integrations/raftstore/test_bootstrap.rs
index e994b7cfc68..dd044eb7369 100644
--- a/tests/integrations/raftstore/test_bootstrap.rs
+++ b/tests/integrations/raftstore/test_bootstrap.rs
@@ -122,7 +122,7 @@ fn test_node_bootstrap_with_prepared_data() {
         importer,
         split_check_scheduler,
         AutoSplitController::default(),
-        ConcurrencyManager::new(1.into()),
+        ConcurrencyManager::new_for_test(1.into()),
         CollectorRegHandle::new_for_test(),
         None,
         DiskCheckRunner::dummy(),
diff --git a/tests/integrations/resource_metering/test_cpu.rs b/tests/integrations/resource_metering/test_cpu.rs
index 12d6fa4fbe0..b75a27320e2 100644
--- a/tests/integrations/resource_metering/test_cpu.rs
+++ b/tests/integrations/resource_metering/test_cpu.rs
@@ -226,7 +226,7 @@ fn setup_test_suite() -> (TestSuite, Store<RocksEngine>, Endpoint<RocksEngine>)
         &CoprReadPoolConfig::default_for_test(),
         store.get_engine(),
     ));
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     let endpoint = Endpoint::new(
         &Default::default(),
         pool.handle(),
diff --git a/tests/integrations/resource_metering/test_read_keys.rs b/tests/integrations/resource_metering/test_read_keys.rs
index f7a4ef86906..5f63d2cd031 100644
--- a/tests/integrations/resource_metering/test_read_keys.rs
+++ b/tests/integrations/resource_metering/test_read_keys.rs
@@ -222,7 +222,7 @@ fn init_coprocessor_with_data(
         &CoprReadPoolConfig::default_for_test(),
         store.get_engine(),
     ));
-    let cm = ConcurrencyManager::new(1.into());
+    let cm = ConcurrencyManager::new_for_test(1.into());
     Endpoint::new(
         &tikv::server::Config::default(),
         pool.handle(),
diff --git a/tests/integrations/server/kv_service.rs b/tests/integrations/server/kv_service.rs
index 4b1b7492474..a2274f82d63 100644
--- a/tests/integrations/server/kv_service.rs
+++ b/tests/integrations/server/kv_service.rs
@@ -1408,7 +1408,7 @@ fn test_double_run_node() {
             importer,
             split_check_scheduler,
             AutoSplitController::default(),
-            ConcurrencyManager::new(1.into()),
+            ConcurrencyManager::new_for_test(1.into()),
             CollectorRegHandle::new_for_test(),
             None,
             DiskCheckRunner::dummy(),