Repository: tikv/tikv (language: rust)

Issue:
tikv panic for "assertion failed: row.lock.is_none()"
## 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?
["Git Commit Hash:   b7103fab92af010ce709238a2e243dc8ab7eba56"] [thread_id=1]

### What operating system and CPU are you using?
<!-- If you're using Linux, you can run `cat /proc/cpuinfo` -->

### Steps to reproduce
<!-- If possible, provide a recipe for reproducing the error. A complete runnable program is good. -->
bank-tbls-pessimistic-inplace-follower-neterr
txn-mode=pessimistic, follower-read=true
run bank workload, 
injuct network fail
params: 
          --txn-mode=pessimistic --force-reinstall=true --update-in-place=true
          --follower-read=true
          --nemesis=schedules,partition-pd-leader,partition-half,partition-ring
          --read-lock=update --os=image --time-limit=600 --version=master
          --workload=bank-multitable --init-sql='set
          @@tidb_enable_mutation_checker=1, @@tidb_txn_assertion_level=strict,
          @@tidb_constraint_check_in_place_pessimistic=off'
### What did you expect?
pass

### What did happened?
```
[2024/04/11 19:01:19.274 +00:00] [FATAL] [lib.rs:477] ["assertion failed: row.lock.is_none()"] [backtrace="   0: tikv_util::set_panic_hook::{{closure}}
             at /workspace/source/tikv/components/tikv_util/src/lib.rs:476:18
   1: <alloc::boxed::Box<F,A> as core::ops::function::Fn<Args>>::call
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2029:9
      std::panicking::rust_panic_with_hook
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:783:13
   2: std::panicking::begin_panic_handler::{{closure}}
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:649:13
   3: std::sys_common::backtrace::__rust_end_short_backtrace
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:171:18
   4: rust_begin_unwind
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:645:5
   5: core::panicking::panic_fmt
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:72:14
   6: core::panicking::panic
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:144:5
   7: resolved_ts::cmd::group_row_changes
             at /workspace/source/tikv/components/resolved_ts/src/cmd.rs:224:25
      resolved_ts::cmd::ChangeLog::encode_change_log::{{closure}}
             at /workspace/source/tikv/components/resolved_ts/src/cmd.rs:62:57
      core::iter::adapters::map::map_fold::{{closure}}
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/adapters/map.rs:85:28
      core::iter::traits::iterator::Iterator::fold
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:2640:21
      <core::iter::adapters::map::Map<I,F> as core::iter::traits::iterator::Iterator>::fold
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/adapters/map.rs:125:9
      core::iter::traits::iterator::Iterator::for_each
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:858:9
      alloc::vec::Vec<T,A>::extend_trusted
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/mod.rs:2923:17
      <alloc::vec::Vec<T,A> as alloc::vec::spec_extend::SpecExtend<T,I>>::spec_extend
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/spec_extend.rs:26:9
      <alloc::vec::Vec<T> as alloc::vec::spec_from_iter_nested::SpecFromIterNested<T,I>>::from_iter
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/spec_from_iter_nested.rs:62:9
      alloc::vec::in_place_collect::<impl alloc::vec::spec_from_iter::SpecFromIter<T,I> for alloc::vec::Vec<T>>::from_iter
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/in_place_collect.rs:233:20
      <alloc::vec::Vec<T> as core::iter::traits::collect::FromIterator<T>>::from_iter
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/mod.rs:2791:9
      core::iter::traits::iterator::Iterator::collect
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:2054:9
      resolved_ts::cmd::ChangeLog::encode_change_log
             at /workspace/source/tikv/components/resolved_ts/src/cmd.rs:76:14
   8: resolved_ts::endpoint::Endpoint<T,E,S>::handle_change_log
             at /workspace/source/tikv/components/resolved_ts/src/endpoint.rs:883:32
      <resolved_ts::endpoint::Endpoint<T,E,S> as tikv_util::worker::pool::Runnable>::run
             at /workspace/source/tikv/components/resolved_ts/src/endpoint.rs:1132:46
      tikv_util::worker::pool::Worker::start_with_timer_impl::{{closure}}
             at /workspace/source/tikv/components/tikv_util/src/worker/pool.rs:517:25
   9: <tracker::tls::TrackedFuture<F> as core::future::future::Future>::poll::{{closure}}
             at /workspace/source/tikv/components/tracker/src/tls.rs:64:23
      std::thread::local::LocalKey<T>::try_with
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:270:16
      std::thread::local::LocalKey<T>::with
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:246:9
      <tracker::tls::TrackedFuture<F> as core::future::future::Future>::poll
             at /workspace/source/tikv/components/tracker/src/tls.rs:62:27
      <futures_util::future::future::map::Map<Fut,F> as core::future::future::Future>::poll
             at /workspace/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.15/src/future/future/map.rs:55:37
      <futures_util::future::future::Map<Fut,F> as core::future::future::Future>::poll
             at /workspace/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.15/src/lib.rs:93:13
      yatp::task::future::RawTask<F>::poll
             at /workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:59:9
  10: yatp::task::future::TaskCell::poll
             at /workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:103:9
      <yatp::task::future::Runner as yatp::pool::runner::Runner>::handle
             at /workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:387:20
  11: <tikv_util::yatp_pool::YatpPoolRunner<T> as yatp::pool::runner::Runner>::handle
             at /workspace/source/tikv/components/tikv_util/src/yatp_pool/mod.rs:198:24
      yatp::pool::worker::WorkerThread<T,R>::run
             at /workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/worker.rs:48:13
      yatp::pool::builder::LazyBuilder<T>::build::{{closure}}
             at /workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/builder.rs:114:25
      std::sys_common::backtrace::__rust_begin_short_backtrace
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:155:18
  12: std::thread::Builder::spawn_unchecked_::{{closure}}::{{closure}}
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:529:17
      <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panic/unwind_safe.rs:272:9
      std::panicking::try::do_call
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:552:40
      std::panicking::try
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:516:19
      std::panic::catch_unwind
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panic.rs:142:14
      std::thread::Builder::spawn_unchecked_::{{closure}}
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:528:30
      core::ops::function::FnOnce::call_once{{vtable.shim}}
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ops/function.rs:250:5
  13: <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9
      <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9
      std::sys::unix::thread::Thread::new::thread_start
             at /root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys/unix/thread.rs:108:17
  14: start_thread
  15: clone
"] [location=components/resolved_ts/src/cmd.rs:224] [thread_name=resolved-ts-0] [thread_id=75]

```


Discussion / hints:
~~It seems to be related to the pipelined dml implementation, which breaks the constraint `Prewrite Lock` would be written at most one time.~~ Need more investigations.

Verfied the `pipelined-dml` is not enabled in this case
@cfzjywxk @seiya-annie Did you tested "follwer-read=false" case? Is this because follower-read?
> @cfzjywxk @seiya-annie Did you tested "follwer-read=false" case? Is this because follower-read?

@zhangjinpeng87  

This issue has also occurred in non-follower-read Jepsen test jobs. The panic occurs during the observation of the apply result process in `resolved-ts`. It appears that the `Requests` of `RaftCmdRequest` involving a lock write for the same key twice which does not comply with the transaction protocol, indicating a correctness issue.

Currently, we have added logging before the assertion. After reproducing the issue, we can further examine the logs to determine which transaction's write operation caused this situation.
A recent reproducing, with more logs:

"there is already 
lock=Put(None, [83, 38, 116, 128, 0, 0, 0, 0, 0, 0, 198, 95, 114, 128, 0, 0, 0, 0, 0, 0, 3, 129, 128, 224, 206, 157, 222, 153, 159, 6, 154, 214, 2, 102, 6, 62, 102, 241, 217, 216, 0, 1, 99, 6, 62, 102, 241, 217, 216, 0, 2, 108, 6, 62, 102, 241, 118, 160, 0, 33, 2]) 

on 

row=RowChange { write: None, lock: Some(Put(None, [83, 38, 116, 128, 0, 0, 0, 0, 0, 0, 198, 95, 114, 128, 0, 0, 0, 0, 0, 0, 3, 129, 128, 224, 206, 157, 222, 153, 159, 6, 154, 214, 2, 102, 6, 62, 102, 241, 217, 216, 0, 1, 99, 6, 62, 102, 241, 217, 216, 0, 2, 108, 6, 62, 102, 241, 118, 160, 0, 33, 2])), default: None }"

which is

Lock { lock_type: Pessimistic, primary_key: 7480000000000000C65F728000000000000003, start_ts: TimeStamp(449910201711591425), ttl: 43802, short_value: , for_update_ts: TimeStamp(449910201711591425), txn_size: 0, min_commit_ts: TimeStamp(449910201711591426), use_async_commit: false, secondaries: [], rollback_ts: [], last_change: Exist { last_change_ts: TimeStamp(449910200046977057), estimated_versions_to_last_change: 2 }, txn_source: 0, is_locked_with_conflict: false, generation: 0 }

The related query is 
```
insert into txn6 (id, sk, val) values (3, 3, 13) on duplicate key update val = 13; // with pessimistic mode
```


But the to be inserted `key` and `value` are not printted in the last log enhancement PR..

It's still not clear why there could be duplicate keys in a single apply batch.
After some investigations, one of the possbility is that pessimistic and prewrite locks with the same key are applied in a same apply batch and it is observed by the `resolved-ts` component, then the assertion is break.

Consider a leader transfer situation
```
region with p1, p2, p3, p1 is the leader peer
1. Pessimistic lock k0 on p1, use in-memory lock, return success
2. P1 attempts to transfer the leader to p3. According to the in-memory lock requirement, before officially initiating, propose an in-memory lock. Here, the lock has not been deleted yet. Propose pessimistic lock to raft log, assuming index = 10. Afterwards, propose transfer leader command index = 11.
3. User prewrite k1 on p1, propose prewrite k0, index = 12 
```

In this case
```
In this case,
log index = 10 pessimistic lock
log index = 11 transfer leader admin command
log index = 12 prewrite lock
```

When the follower peer P3 apply logs, the pessimistic lock and prewrite locks could be consumed in a same apply batch, like

apply.rs:646: [INFO] [for debug] on store_id=3 cmd_batch=[CmdBatch { level: None, cdc_id: ObserveId(0), rts_id: ObserveId(0), pitr_id: ObserveId(0), region_id: 1, cmds: [**Cmd { index: 10, term: 6, request: header { region_id: 1 peer { id: 1 store_id: 1 } region_epoch { conf_ver: 3 version: 1 } } requests { cmd_type: Put put { cf: "lock" key: 6B30000000000000F9 value: 53046B300A0066000000000000000A6C000000000000000001 }** }, response: header { current_term: 6 } }, Cmd { index: 11, term: 6, request: header { region_id: 1 peer { id: 1 store_id: 1 } region_epoch { conf_ver: 3 version: 1 } flags: 4 } admin_request { cmd_type: TransferLeader transfer_leader { peer { id: 3 store_id: 3 } } }, response: header { current_term: 6 } admin_response { cmd_type: TransferLeader } }, **Cmd { index: 12, term: 6, request: header { region_id: 1 peer { id: 1 store_id: 1 } region_epoch { conf_ver: 3 version: 1 } term: 6 } requests { cmd_type: Put put { cf: "lock" key: 6B30000000000000F9 value: 50046B300AB8177602763066000000000000000A63000000000000000B } }**, response: header { current_term: 6 } }] }], thread_id: 486


It's still not clear how the `resolved-ts` could observe such situations, as the region would be registerred to the `resolved-ts` after the new leader has applied to its current term.

> After some investigations, one of the possbility is that pessimistic and prewrite locks with the same key are applied in a same apply batch and it is observed by the `resolved-ts` component, then the assertion is break.
> 
> Consider a leader transfer situation
> 
> ```
> region with p1, p2, p3, p1 is the leader peer
> 1. Pessimistic lock k0 on p1, use in-memory lock, return success
> 2. P1 attempts to transfer the leader to p3. According to the in-memory lock requirement, before officially initiating, propose an in-memory lock. Here, the lock has not been deleted yet. Propose pessimistic lock to raft log, assuming index = 10. Afterwards, propose transfer leader command index = 11.
> 3. User prewrite k1 on p1, propose prewrite k0, index = 12 
> ```
> 
> In this case
> 
> ```
> In this case,
> log index = 10 pessimistic lock
> log index = 11 transfer leader admin command
> log index = 12 prewrite lock
> ```
> 
> When the follower peer P3 apply logs, the pessimistic lock and prewrite locks could be consumed in a same apply batch, like
> 
> apply.rs:646: [INFO] [for debug] on store_id=3 cmd_batch=[CmdBatch { level: None, cdc_id: ObserveId(0), rts_id: ObserveId(0), pitr_id: ObserveId(0), region_id: 1, cmds: [**Cmd { index: 10, term: 6, request: header { region_id: 1 peer { id: 1 store_id: 1 } region_epoch { conf_ver: 3 version: 1 } } requests { cmd_type: Put put { cf: "lock" key: 6B30000000000000F9 value: 53046B300A0066000000000000000A6C000000000000000001 }** }, response: header { current_term: 6 } }, Cmd { index: 11, term: 6, request: header { region_id: 1 peer { id: 1 store_id: 1 } region_epoch { conf_ver: 3 version: 1 } flags: 4 } admin_request { cmd_type: TransferLeader transfer_leader { peer { id: 3 store_id: 3 } } }, response: header { current_term: 6 } admin_response { cmd_type: TransferLeader } }, **Cmd { index: 12, term: 6, request: header { region_id: 1 peer { id: 1 store_id: 1 } region_epoch { conf_ver: 3 version: 1 } term: 6 } requests { cmd_type: Put put { cf: "lock" key: 6B30000000000000F9 value: 50046B300AB8177602763066000000000000000A63000000000000000B } }**, response: header { current_term: 6 } }] }], thread_id: 486
> 
> It's still not clear how the `resolved-ts` could observe such situations, as the region would be registerred to the `resolved-ts` after the new leader has applied to its current term.

But the situation here is duplicate keys in different logs, while the panic happens when there are duplicate keys within a same log which is different..
A new reproducing with more logs, two pessimsitic locks on the key are handled in a same `RaftCmdRequest`, which is unexepected.

---



[2024/07/20 06:29:42.467 +00:00] [ERROR] [cmd.rs:243] ["there is already row=RowChange { write: None, lock: Some(Put(ts:None, value:53267480000000000000985F7280000000000000298B8090ECA5EACEA106F7DF026606433B525D84000F6306433B525D84000C6C06433B525D84000F0146)), default: None } with same key processing key=7480000000000000FF985F728000000000FF0000290000000000FA value=53267480000000000000985F728000000000000029978090ECA5EACEA106FADA026606433B525D8400176306433B525D8400186C06433B525D84000F01"] [thread_id=99]

existing lock=Lock { lock_type: Pessimistic, primary_key: 7480000000000000985F728000000000000029, start_ts: TimeStamp(**451269612609470475**), ttl: 45047, short_value: , for_update_ts: TimeStamp(451269612609470479), txn_size: 0, min_commit_ts: TimeStamp(451269612609470476), use_async_commit: false, secondaries: [], rollback_ts: [], last_change: Exist { last_change_ts: TimeStamp(451269612609470479), estimated_versions_to_last_change: 1 }, txn_source: 0, is_locked_with_conflict: true, generation: 0 }

new lock=Lock { lock_type: Pessimistic, primary_key: 7480000000000000985F728000000000000029, start_ts: TimeStamp(**451269612609470487**), ttl: 44410, short_value: , for_update_ts: TimeStamp(451269612609470487), txn_size: 0, min_commit_ts: TimeStamp(451269612609470488), use_async_commit: false, secondaries: [], rollback_ts: [], last_change: Exist { last_change_ts: TimeStamp(451269612609470479), estimated_versions_to_last_change: 1 }, txn_source: 0, is_locked_with_conflict: false, generation: 0 }




txn1 451269612609470475
```
# Time: 2024-07-20T06:29:42.471178774Z
# Txn_start_ts: 451269612609470475
# User@Host: root[root] @ 10.200.65.161 [10.200.65.161]
# Conn_ID: 4125097998
# Exec_retry_time: 25.164334311 Exec_retry_count: 1
# Query_time: 25.167019728
# Parse_time: 0.000026797
# Compile_time: 0.000044644
# Rewrite_time: 0.00000323
# Optimize_time: 0.000030689
# Wait_TS: 0.00023222
# LockKeys_time: 25.166502883
# DB: test
# Is_internal: false
# Digest: dd03a9b946aecdfd38bc3013e57e8d0f2a7a2c7b0941559966463bf30b8b4b9c
# Stats: txn6:pseudo[10000;0]
# Num_cop_tasks: 0
# Mem_max: 4096
# Prepared: false
# Plan_from_cache: false
# Plan_from_binding: false
# Has_more_results: false
# KV_total: 25.162398889
# PD_total: 0.000228617
# Backoff_total: 0
# Write_sql_response_total: 0.000035325
# Result_rows: 1
# Succ: true
# IsExplicitTxn: true
# IsSyncStatsFailed: false
# Plan: tidb_decode_plan('4ALIMAkzNl8xCTAJMQl0YWJsZTp0eG42LCBoYW5kbGU6NDEsIGxvY2sJMQl0aW1lOjI1LjJzARMQb3BzOjMBCSRja19rZXlzOiB7CSFALjQybXMsIHJlZ2lvbjoxLCAFHjQxLCBzbG93ZXN0X3JwYwEtJG90YWw6IDAuMDABUQkvWF9pZDogNTIsIHN0b3JlOiBub2RlLTEuBQeQcGVlci5qZXBzZW4tdHBzLTc2MDU1MDQtMS02Mjc6MjAxNjAsIAGDGF9kZXRhaWwBZCBpa3Zfd2FsbF8BGDg6IDEuMzNtc30sIHNjYW4ZJyxnZXRfc25hcHNob3QNKgwzNcK1AcVYb2Nrc2RiOiB7YmxvY2s6IHt9fX0sIH0N+DBycGM6Mi4zNjg5MzZtAS9McGNfY291bnQ6MX0JTi9BCU4vQQo=')
# Plan_digest: 0f8e0de76ece9905c88e8fd971f9269e1d6fcdaf898bb4f2d05c6d0aa66950b0
# Binary_plan: tidb_decode_binary_plan('iQNECoQDCgtQb2ludF9HZXRfMSkABQHQ8D8wATgBQAFKDgoMCgR0ZXN0EgR0eG42Ug9oYW5kbGU6NDEsIGxvY2taE3RpbWU6MjUuMnMBEhxvcHM6M2KZAgEcHF9rZXlzOiB7CSJALjQybXMsIHJlZ2lvbjoxLCAFHjQxLCBzbG93ZXN0X3JwYwEtJG90YWw6IDAuMDABUgkvWF9pZDogNTIsIHN0b3JlOiBub2RlLTEuBQeQcGVlci5qZXBzZW4tdHBzLTc2MDU1MDQtMS02Mjc6MjAxNjAsIAGDGF9kZXRhaWwBZCBpa3Zfd2FsbF8BGDg6IDEuMzNtc30sIHNjYW4ZJyxnZXRfc25hcHNob3QNKgwzNcK1AcVYb2Nrc2RiOiB7YmxvY2s6IHt9fX0sIH0hAgRjawXKIDIuMzY4OTM2bQEvMHBjX2NvdW50OjF9cP8RATQBeP///////////wEYAQ==')
# Resource_group: default
# Request_unit_write: 142.79999999999993
select val from txn6 where id = 41 FOR UPDATE;
```

txn2 451269612609470487 
```
# Time: 2024-07-20T06:29:42.467100628Z
# Txn_start_ts: 451269612609470487
# User@Host: root[root] @ 10.200.65.161 [10.200.65.161]
# Conn_ID: 3152019494
# Query_time: 25.141969218
# Parse_time: 0.000066793
# Compile_time: 0.000188155
# Rewrite_time: 0.000048971
# Optimize_time: 0.000062031
# Wait_TS: 0.000682948
# LockKeys_time: 25.139729056
# DB: test
# Is_internal: false
# Digest: a51559580d4ab4b95aae2fdd415c78296f0f772e7b6d9b1539bb7d1c16e5d79e
# Num_cop_tasks: 0
# Mem_max: 8408
# Prepared: false
# Plan_from_cache: false
# Plan_from_binding: false
# Has_more_results: false
# KV_total: 25.137000108
# PD_total: 0.00067564
# Backoff_total: 0
# Write_sql_response_total: 0
# Result_rows: 0
# Warnings: [{"Level":"Warning","Message":"Pipelined DML can only be used in autocommit mode. Fallback to standard mode"}]
# Succ: true
# IsExplicitTxn: true
# IsSyncStatsFailed: false
# Plan: tidb_decode_plan('zwXwTDAJMjdfMQkwCTAJTi9BCTAJdGltZToyNS4xcywgbG9vcHM6MiwgcHJlcGFyZTogMjkuNMK1cywgY2hlY2tfaW5zZXJ0OiB7dG90YWxfBTwMIDkzNg0lCG1lbQ0jAF8JGww4My41BT84cHJlZmV0Y2g6IDg1Mi45BRRIcnBjOntCYXRjaEdldDp7bnVtXwETCDEsIB1eKDc4OS42wrVzfSwgAVMYX2RldGFpbBWCLGt2X3JlYWRfd2FsbA10EDM0MC4xBWEMdGlrdjIaAAQ1OAUaGH0sIHNjYW4+TAA0cHJvY2Vzc19rZXlzOiAVhy4XABxfc2l6ZTogNBUdFSwoZ2V0X3NuYXBzaG8R/RAyOTkuMgWJSHJvY2tzZGI6IHtibG9jazoge30BAQQsIAEODUQAewHfMYkUcmVnaW9uIQsFYSgxLCByZXNvbHZlXwEyNDoxOG1zLCBzbG93ZXN0JTUEIHslNBw6IDAuNzMxcwEwAUIsX2lkOiA1Miwgc3RvIc8Ybm9kZS00LgUHiHBlZXIuamVwc2VuLXRwcy03NjA1NTA0LTEtNjI3OjIwMTYwQnQBOlEBGDczMC4ybXNCUAFKDQEEMTNuDAEELCAxDUECQY4cMjEzNzU2NjkBy2hwY19jb3VudDoyNX0JMjE2IEJ5dGVzCU4vQQo=')
# Binary_plan: tidb_decode_binary_plan('2QXwRgrUBQoISW5zZXJ0XzE4AUABUgNOL0FaE3RpbWU6MjUuMXMsIGxvb3BzOjJi5gJwcmVwYXJlOiAyOS40wrVzLCBjaGVja19pBUEgOiB7dG90YWxfBT0MIDkzNg0lEG1lbV9pBSMNGww4My41BT84cHJlZmV0Y2g6IDg1Mi45BRRIcnBjOntCYXRjaEdldDp7bnVtXwETCDEsIB1eKDc4OS42wrVzfSwgAW4YX2RldGFpbBWCKGt2X3JlYWRfd2FsEY8QMzQwLjEFYQx0aWt2MhoABDU4BRoYfSwgc2Nhbj5MADRwcm9jZXNzX2tleXM6IBWHLhcAHF9zaXplOiA0FR0VLChnZXRfc25hcHNobxH9EDI5OS4yBYlIcm9ja3NkYjoge2Jsb2NrOiB7fQEBCGKyAgEPDUUAewHgMYsUcmVnaW9uIQwFYigxLCByZXNvbHZlXwEyNDoxOG1zLCBzbG93ZXN0JTYtkxw6IDAuNzMxcwEwAUIsX2lkOiA1Miwgc3RvIdAYbm9kZS00LgUHiHBlZXIuamVwc2VuLXRwcy03NjA1NTA0LTEtNjI3OjIwMTYwQnUBOlIBGDczMC4ybXNCUQFKDgEEMTNuDQEQLCB9LCAB2wXNQZAcMjEzNzU2NjkBy2xwY19jb3VudDoyNX1w2AF4////////////ARgB')
# Resource_group: default
# Request_unit_read: 0.4756256103515625
# Request_unit_write: 127.49999999999994
insert into txn6 (id, sk, val) values (41, 41, 18) on duplicate key update val = 18;
```
A new reproducing with tikv version https://github.com/tikv/tikv/commit/3bd8c24d1a1eb277877557b35d3b175330e7010e: 
```
[2024/11/23 04:14:03.759 +00:00] [ERROR] [[cmd.rs:246](http://cmd.rs:246/)] ["there is already row=RowChange { write: None, lock: Some(Put(ts:None, value:50267480000000000000185F7280000000000DADE7ED85C0B7849DD7A606B917763F080208A4C221080408000806080C08080800080A0800080C08E6BA04080E08000810099F80A0FCB4BECFA6060814080008160800081805800000000000000074000000000000000463064D5CE846F002EE)), default: None } with same key processing key=7480000000000000FF185F728000000000FF0DADE70000000000FA value=50267480000000000000185F7280000000000DADE7ED85C0B7849DD7A606B917763F080208A4C221080408000806080C08080800080A0800080C08E6BA04080E08000810099F80A0FCB4BECFA6060814080008160800081805800000000000000074000000000000000463064D5CE846F002EE"] [thread_id=99]
```


```
existing lock=Lock { lock_type: Put, primary_key: 7480000000000000185F7280000000000DADE7, start_ts: TimeStamp(454121290126590701), ttl: 3001, short_value: 080208A4C221080408000806080C08080800080A0800080C08E6BA04080E08000810099F80A0FCB4BECFA60608140800081608000818058000000000000000, for_update_ts: TimeStamp(0), txn_size: 4, min_commit_ts: TimeStamp(454121290126590702), use_async_commit: false, secondaries: [], rollback_ts: [], last_change: Unknown, txn_source: 0, is_locked_with_conflict: false, generation: 0 }
newwwwww lock=Lock { lock_type: Put, primary_key: 7480000000000000185F7280000000000DADE7, start_ts: TimeStamp(454121290126590701), ttl: 3001, short_value: 080208A4C221080408000806080C08080800080A0800080C08E6BA04080E08000810099F80A0FCB4BECFA60608140800081608000818058000000000000000, for_update_ts: TimeStamp(0), txn_size: 4, min_commit_ts: TimeStamp(454121290126590702), use_async_commit: false, secondaries: [], rollback_ts: [], last_change: Unknown, txn_source: 0, is_locked_with_conflict: false, generation: 0 }
same=true
```

From the log before assertion failure, two identical key-value locks appeared in the same raft command request.
I found a similar panic in jepsen test, duplicated locks were observed by the resolved_ts module.

```log
store/TiDB master bank auto-retry auto-retry-limit :default update-in-place select FOR UPDATE txn-mode pessimistic isolation :repeatable-read nemesis failpoint/20241205T084444.000Z/node-2.node-peer.jepsen-tps-7694235-1-919/kv.log:
[2024/12/05 08:52:23.049 +00:00] [FATAL] [lib.rs:480] ["assertion failed: occupied_entry.get().lock.is_none()"] [backtrace="   0: tikv_util::set_panic_hook::{{closure}}\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/tikv_util/src/lib.rs:479:18\n   1: <alloc::boxed::Box<F,A> as core::ops::function::Fn<Args>>::call\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2029:9\n      std::panicking::rust_panic_with_hook\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:783:13\n   2: std::panicking::begin_panic_handler::{{closure}}\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:649:13\n   3: std::sys_common::backtrace::__rust_end_short_backtrace\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:171:18\n   4: rust_begin_unwind\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:645:5\n   5: core::panicking::panic_fmt\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:72:14\n   6: core::panicking::panic\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:144:5\n   7: resolved_ts::cmd::group_row_changes\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/resolved_ts/src/cmd.rs:253:33\n      resolved_ts::cmd::ChangeLog::encode_change_log::{{closure}}\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/resolved_ts/src/cmd.rs:65:57\n      core::iter::adapters::map::map_fold::{{closure}}\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/adapters/map.rs:85:28\n      core::iter::traits::iterator::Iterator::fold\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:2640:21\n      <core::iter::adapters::map::Map<I,F> as core::iter::traits::iterator::Iterator>::fold\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/adapters/map.rs:125:9\n      core::iter::traits::iterator::Iterator::for_each\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:858:9\n      alloc::vec::Vec<T,A>::extend_trusted\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/mod.rs:2923:17\n      <alloc::vec::Vec<T,A> as alloc::vec::spec_extend::SpecExtend<T,I>>::spec_extend\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/spec_extend.rs:26:9\n      <alloc::vec::Vec<T> as alloc::vec::spec_from_iter_nested::SpecFromIterNested<T,I>>::from_iter\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/spec_from_iter_nested.rs:62:9\n      alloc::vec::in_place_collect::<impl alloc::vec::spec_from_iter::SpecFromIter<T,I> for alloc::vec::Vec<T>>::from_iter\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/in_place_collect.rs:233:20\n      <alloc::vec::Vec<T> as core::iter::traits::collect::FromIterator<T>>::from_iter\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/vec/mod.rs:2791:9\n      core::iter::traits::iterator::Iterator::collect\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:2054:9\n      resolved_ts::cmd::ChangeLog::encode_change_log\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/resolved_ts/src/cmd.rs:79:14\n   8: resolved_ts::endpoint::Endpoint<T,E,S>::handle_change_log\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/resolved_ts/src/endpoint.rs:919:32\n      <resolved_ts::endpoint::Endpoint<T,E,S> as tikv_util::worker::pool::Runnable>::run\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/resolved_ts/src/endpoint.rs:1168:46\n      tikv_util::worker::pool::Worker::start_with_timer_impl::{{closure}}\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/tikv_util/src/worker/pool.rs:532:25\n   9: <tracker::tls::TrackedFuture<F> as core::future::future::Future>::poll::{{closure}}\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/tracker/src/tls.rs:64:23\n      std::thread::local::LocalKey<T>::try_with\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:270:16\n      std::thread::local::LocalKey<T>::with\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:246:9\n      <tracker::tls::TrackedFuture<F> as core::future::future::Future>::poll\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/tracker/src/tls.rs:62:27\n      <futures_util::future::future::map::Map<Fut,F> as core::future::future::Future>::poll\n             at /var/cache/cargohome/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.31/src/future/future/map.rs:55:37\n      <futures_util::future::future::Map<Fut,F> as core::future::future::Future>::poll\n             at /var/cache/cargohome/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.31/src/lib.rs:86:13\n      yatp::task::future::RawTask<F>::poll\n             at /var/cache/cargohome/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:59:9\n  10: yatp::task::future::TaskCell::poll\n             at /var/cache/cargohome/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:103:9\n      <yatp::task::future::Runner as yatp::pool::runner::Runner>::handle\n             at /var/cache/cargohome/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:387:20\n  11: <tikv_util::yatp_pool::YatpPoolRunner<T> as yatp::pool::runner::Runner>::handle\n             at /home/jenkins/agent/workspace/build-common/go/src/github.com/pingcap/tikv/components/tikv_util/src/yatp_pool/mod.rs:199:24\n      yatp::pool::worker::WorkerThread<T,R>::run\n             at /var/cache/cargohome/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/worker.rs:48:13\n      yatp::pool::builder::LazyBuilder<T>::build::{{closure}}\n             at /var/cache/cargohome/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/builder.rs:114:25\n      std::sys_common::backtrace::__rust_begin_short_backtrace\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:155:18\n  12: std::thread::Builder::spawn_unchecked_::{{closure}}::{{closure}}\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:529:17\n      <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panic/unwind_safe.rs:272:9\n      std::panicking::try::do_call\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:552:40\n      std::panicking::try\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:516:19\n      std::panic::catch_unwind\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panic.rs:142:14\n      std::thread::Builder::spawn_unchecked_::{{closure}}\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:528:30\n      core::ops::function::FnOnce::call_once{{vtable.shim}}\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ops/function.rs:250:5\n  13: <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9\n      <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9\n      std::sys::unix::thread::Thread::new::thread_start\n             at /rust/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys/unix/thread.rs:108:17\n  14: start_thread\n  15: clone\n"] [location=components/resolved_ts/src/cmd.rs:253] [thread_name=resolved-ts-0] [thread_id=106]
```

https://tcms.pingcap.net/dashboard/executions/plan/7694235

BTW, the TiKV includes debug logs that trace write operations, should be helpful for debugging.
May be a new reproduction in #18108 
From the new reproduction in https://github.com/tikv/tikv/issues/18108, 

The root cause seems to be that during the transfer leader process(or other region changes like split/merge), the in-memory pessimistic lock and storage may contain the same key lock, and the current `scan lock` operation does not handle this situation properly.

When the scan result is used to trigger some releasing lock commands, there could be duplicate entry writes in a same write command.


Fix patch:
diff --git a/components/raftstore/src/store/fsm/peer.rs b/components/raftstore/src/store/fsm/peer.rs
index 54cef0ab58f..1f09e67edad 100644
--- a/components/raftstore/src/store/fsm/peer.rs
+++ b/components/raftstore/src/store/fsm/peer.rs
@@ -3820,6 +3820,10 @@ where
                 None => {
                     self.propose_pending_batch_raft_command();
                     if self.propose_locks_before_transfer_leader(msg) {
+                        fail_point!(
+                            "finish_proposing_transfer_cmd_after_proposing_locks",
+                            |_| {}
+                        );
                         // If some pessimistic locks are just proposed, we propose another
                         // TransferLeader command instead of transferring leader immediately.
                         info!("propose transfer leader command";
diff --git a/src/storage/mvcc/reader/reader.rs b/src/storage/mvcc/reader/reader.rs
index 454250737be..5e26f964521 100644
--- a/src/storage/mvcc/reader/reader.rs
+++ b/src/storage/mvcc/reader/reader.rs
@@ -350,30 +350,40 @@ impl<S: EngineSnapshot> MvccReader<S> {
             Ok(None)
         };
 
-        let mut locks = Vec::with_capacity(limit.min(memory_locks.len()));
+        let mut locks: Vec<(Key, Lock)> = Vec::with_capacity(limit.min(memory_locks.len()));
         let mut memory_iter = memory_locks.into_iter();
         let mut memory_pair = memory_iter.next();
         let mut storage_pair = next_pair_from_storage()?;
         let has_remain = loop {
-            match (memory_pair.as_ref(), storage_pair.as_ref()) {
+            let next_key = match (memory_pair.as_ref(), storage_pair.as_ref()) {
                 (Some((memory_key, _)), Some((storage_key, _))) => {
                     if storage_key <= memory_key {
-                        locks.push(storage_pair.take().unwrap());
+                        let next_key = storage_pair.take().unwrap();
                         storage_pair = next_pair_from_storage()?;
+                        next_key
                     } else {
-                        locks.push(memory_pair.take().unwrap());
+                        let next_key = memory_pair.take().unwrap();
                         memory_pair = memory_iter.next();
+                        next_key
                     }
                 }
                 (Some(_), None) => {
-                    locks.push(memory_pair.take().unwrap());
+                    let next_key = memory_pair.take().unwrap();
                     memory_pair = memory_iter.next();
+                    next_key
                 }
                 (None, Some(_)) => {
-                    locks.push(storage_pair.take().unwrap());
+                    let next_key = storage_pair.take().unwrap();
                     storage_pair = next_pair_from_storage()?;
+                    next_key
                 }
                 (None, None) => break memory_has_remain,
+            };
+            // The same key could exist in both memory and storage when there is ongoing
+            // leader transfer, split or merge on this region. In this case, duplicated
+            // keys should be ignored.
+            if locks.is_empty() || locks.last().unwrap().0 != next_key.0 {
+                locks.push(next_key);
             }
             if limit > 0 && locks.len() >= limit {
                 break memory_pair.is_some() || storage_pair.is_some() || memory_has_remain;


Test patch:
diff --git a/components/test_raftstore/src/cluster.rs b/components/test_raftstore/src/cluster.rs
index 0a165ec0528..374533b427c 100644
--- a/components/test_raftstore/src/cluster.rs
+++ b/components/test_raftstore/src/cluster.rs
@@ -1465,6 +1465,17 @@ impl<T: Simulator> Cluster<T> {
             .unwrap()
     }
 
+    pub fn try_transfer_leader_with_timeout(
+        &mut self,
+        region_id: u64,
+        leader: metapb::Peer,
+        timeout: Duration,
+    ) -> Result<RaftCmdResponse> {
+        let epoch = self.get_region_epoch(region_id);
+        let transfer_leader = new_admin_request(region_id, &epoch, new_transfer_leader_cmd(leader));
+        self.call_command_on_leader(transfer_leader, timeout)
+    }
+
     pub fn get_snap_dir(&self, node_id: u64) -> String {
         self.sim.rl().get_snap_dir(node_id)
     }
diff --git a/components/test_raftstore/src/util.rs b/components/test_raftstore/src/util.rs
index f6f513b767b..1cd4a17d730 100644
--- a/components/test_raftstore/src/util.rs
+++ b/components/test_raftstore/src/util.rs
@@ -1305,8 +1305,13 @@ pub fn kv_pessimistic_lock_with_ttl(
 }
 
 pub fn must_kv_pessimistic_lock(client: &TikvClient, ctx: Context, key: Vec<u8>, ts: u64) {
-    let resp = kv_pessimistic_lock(client, ctx, vec![key], ts, ts, false);
-    assert!(!resp.has_region_error(), "{:?}", resp.get_region_error());
+    let resp = kv_pessimistic_lock(client, ctx.clone(), vec![key], ts, ts, false);
+    assert!(
+        !resp.has_region_error(),
+        "{:?}, ctx:{:?}",
+        resp.get_region_error(),
+        ctx
+    );
     assert!(resp.errors.is_empty(), "{:?}", resp.get_errors());
 }
 
diff --git a/tests/failpoints/cases/test_kv_service.rs b/tests/failpoints/cases/test_kv_service.rs
index c8777282787..c51b6232ba6 100644
--- a/tests/failpoints/cases/test_kv_service.rs
+++ b/tests/failpoints/cases/test_kv_service.rs
@@ -2,18 +2,20 @@
 
 use std::{sync::Arc, time::Duration};
 
+use engine_traits::{Peekable, CF_LOCK};
 use grpcio::{ChannelBuilder, Environment};
 use kvproto::{
     kvrpcpb::{PrewriteRequestPessimisticAction::SkipPessimisticCheck, *},
     tikvpb::TikvClient,
 };
 use test_raftstore::{
-    configure_for_lease_read, must_kv_commit, must_kv_have_locks, must_kv_prewrite,
-    must_kv_prewrite_with, must_new_cluster_mul, new_server_cluster, try_kv_prewrite_with,
-    try_kv_prewrite_with_impl,
+    configure_for_lease_read, must_kv_commit, must_kv_have_locks, must_kv_pessimistic_lock,
+    must_kv_prewrite, must_kv_prewrite_with, must_new_cluster_and_kv_client_mul,
+    must_new_cluster_mul, new_server_cluster, try_kv_prewrite_with, try_kv_prewrite_with_impl,
 };
 use test_raftstore_macro::test_case;
-use tikv_util::{config::ReadableDuration, HandyRwLock};
+use tikv_util::{config::ReadableDuration, store::new_peer, HandyRwLock};
+use txn_types::Key;
 
 #[test_case(test_raftstore::must_new_cluster_and_kv_client)]
 #[test_case(test_raftstore_v2::must_new_cluster_and_kv_client)]
@@ -270,3 +272,101 @@ fn test_storage_do_not_update_txn_status_cache_on_write_error() {
     must_kv_have_locks(&client, ctx, 29, b"k2", b"k3", &[(b"k2", Op::Put, 20, 20)]);
     fail::remove(cache_hit_fp);
 }
+
+#[test]
+fn test_scan_locks_with_in_progress_transfer_leader() {
+    let (mut cluster, _, mut ctx) = must_new_cluster_and_kv_client_mul(3);
+    cluster.pd_client.disable_default_operator();
+
+    cluster.must_transfer_leader(1, new_peer(1, 1));
+    let leader_peer = cluster.leader_of_region(1).unwrap();
+    ctx.set_peer(leader_peer.clone());
+    let k1 = b"k1";
+    let k2 = b"k2";
+    let leader_region = cluster.get_region(k1);
+    ctx.set_region_epoch(leader_region.get_region_epoch().clone());
+    let env = Arc::new(Environment::new(1));
+    let channel =
+        ChannelBuilder::new(env).connect(&cluster.sim.rl().get_addr(leader_peer.get_store_id()));
+    let client = TikvClient::new(channel);
+
+    // Create both pessimistic locks.
+    let start_ts = 10;
+    must_kv_pessimistic_lock(&client, ctx.clone(), k1.to_vec(), start_ts);
+    must_kv_pessimistic_lock(&client, ctx.clone(), k2.to_vec(), start_ts);
+
+    // Ensure the pessimistic locks are written to the memory but not the storage.
+    let engine = cluster.get_engine(leader_peer.get_store_id());
+    let cf_res = engine
+        .get_value_cf(
+            CF_LOCK,
+            keys::data_key(Key::from_raw(k1).as_encoded()).as_slice(),
+        )
+        .unwrap();
+    assert!(cf_res.is_none());
+    let cf_res = engine
+        .get_value_cf(
+            CF_LOCK,
+            keys::data_key(Key::from_raw(k2).as_encoded()).as_slice(),
+        )
+        .unwrap();
+    assert!(cf_res.is_none());
+
+    let mut scan_lock_req = ScanLockRequest::default();
+    scan_lock_req.set_context(ctx.clone());
+    scan_lock_req.max_version = start_ts + 10;
+    scan_lock_req.limit = 256;
+    let scan_lock_resp = client.kv_scan_lock(&scan_lock_req.clone()).unwrap();
+    assert!(!scan_lock_resp.has_region_error());
+    assert_eq!(scan_lock_resp.get_locks().len(), 2);
+    assert_eq!(scan_lock_resp.locks.to_vec()[0].lock_version, start_ts);
+    assert_eq!(scan_lock_resp.locks.to_vec()[0].key, k1);
+    assert_eq!(scan_lock_resp.locks.to_vec()[1].lock_version, start_ts);
+    assert_eq!(scan_lock_resp.locks.to_vec()[1].key, k2);
+
+    // Propose the transfer leader command but only trigger proposing pessimistic
+    // locks.
+    fail::cfg(
+        "finish_proposing_transfer_cmd_after_proposing_locks",
+        "return",
+    )
+    .unwrap();
+    let _ = cluster.try_transfer_leader_with_timeout(1, new_peer(2, 2), Duration::from_secs(1));
+
+    // Verify locks exist both in memory and storage.
+    let timer = tikv_util::time::Instant::now();
+    let timeout = Duration::from_secs(5);
+    loop {
+        let cf_res = engine
+            .get_value_cf(
+                CF_LOCK,
+                keys::data_key(Key::from_raw(k1).as_encoded()).as_slice(),
+            )
+            .unwrap();
+        if cf_res.is_some() {
+            break;
+        }
+        std::thread::sleep(Duration::from_secs(1));
+        if timer.saturating_elapsed() >= timeout {
+            assert!(cf_res.is_some());
+            break;
+        }
+    }
+    let snapshot = cluster.must_get_snapshot_of_region(1);
+    let txn_ext = snapshot.txn_ext.unwrap();
+    let guard = txn_ext.pessimistic_locks.read();
+    assert!(guard.get(&Key::from_raw(k1)).is_some());
+    assert!(guard.get(&Key::from_raw(k2)).is_some());
+    drop(guard);
+
+    fail::remove("finish_proposing_transfer_cmd_after_proposing_locks");
+
+    // Verify there should be no duplicate locks returned.
+    let scan_lock_resp = client.kv_scan_lock(&scan_lock_req.clone()).unwrap();
+    assert!(!scan_lock_resp.has_region_error());
+    assert_eq!(scan_lock_resp.locks.len(), 2);
+    assert_eq!(scan_lock_resp.locks.to_vec()[0].lock_version, start_ts);
+    assert_eq!(scan_lock_resp.locks.to_vec()[0].key, k1);
+    assert_eq!(scan_lock_resp.locks.to_vec()[1].lock_version, start_ts);
+    assert_eq!(scan_lock_resp.locks.to_vec()[1].key, k2);
+}