Repository: tikv/tikv (language: rust)

Issue:
Leader transfer should be postponed until on-going split is finished
## Development Task

Leader transfer should be postponed util on-going split is finished, otherwise newly splited region can not campaign leader immediately in the following case.

| Time | tikv_10_19_85_6_20163 | tikv_10_19_85_4_20163 |
|--------|--------|--------|
| 14:07:40.495 | region 33977192 starts transfer leader | / |
| 14:07:40.496 | / | region 33977192 splits region 34144801 |
| 14:07:40.496 | / | region 33977192 became leader |
| 14:07:40.502 | region 33977192 became follower | / |
| 14:07:40.502 | region 33977192 splits region 34144801 | / |

It causes TiDB write/read latency spike and TiCDC replication spike.

```log
tikv_10_19_85_4_20163.log:6170:[2022/04/18 14:06:18.991 +08:00] [Info] [raft.rs:783] ["became follower at term 362"] [term=362] [raft_id=34144658] [region_id=33977192]
tikv_10_19_85_4_20163.log:6559:[2022/04/18 14:07:40.495 +08:00] [Info] [raft.rs:1739] ["[term 362] received MsgTimeoutNow from 34144442 and starts an election to get leadership."] [from=34144442] [term=362] [raft_id=34144658] [region_id=33977192]
tikv_10_19_85_4_20163.log:6561:[2022/04/18 14:07:40.495 +08:00] [Info] [raft.rs:807] ["became candidate at term 363"] [term=363] [raft_id=34144658] [region_id=33977192]
tikv_10_19_85_4_20163.log:6566:[2022/04/18 14:07:40.495 +08:00] [Info] [apply.rs:1234] ["execute admin command"] [command="cmd_type: BatchSplit splits { requests { split_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA new_region_id: 34144801 new_peer_ids: 34144802 new_peer_ids: 34144803 new_peer_ids: 34144804 new_peer_ids: 34144805 } right_derive: true }"] [index=4807851] [term=362] [peer_id=34144658] [region_id=33977192]
tikv_10_19_85_4_20163.log:6568:[2022/04/18 14:07:40.496 +08:00] [Info] [peer.rs:2106] ["insert new region"] [region="id: 34144801 start_key: 7480000000000018FF435F7280000004AEFF9A4E430000000000FA end_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA region_epoch { conf_ver: 40837 version: 22856 } peers { id: 34144802 store_id: 11 } peers { id: 34144803 store_id: 14 } peers { id: 34144804 store_id: 1 } peers { id: 34144805 store_id: 9 }"] [region_id=34144801]
tikv_10_19_85_4_20163.log:6575:[2022/04/18 14:07:40.496 +08:00] [Info] [raft.rs:874] ["became leader at term 363"] [term=363] [raft_id=34144658] [region_id=33977192]
tikv_10_19_85_4_20163.log:6675:[2022/04/18 14:07:52.541 +08:00] [Info] [raft.rs:1739] ["[term 6] received MsgTimeoutNow from 34144805 and starts an election to get leadership."] [from=34144805] [term=6] [raft_id=34144804] [region_id=34144801]

tikv_10_19_85_5_20160.log:4737:[2022/04/18 14:07:40.495 +08:00] [Info] [apply.rs:1234] ["execute admin command"] [command="cmd_type: BatchSplit splits { requests { split_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA new_region_id: 34144801 new_peer_ids: 34144802 new_peer_ids: 34144803 new_peer_ids: 34144804 new_peer_ids: 34144805 } right_derive: true }"] [index=4807851] [term=362] [peer_id=34144322] [region_id=33977192]
tikv_10_19_85_5_20160.log:4739:[2022/04/18 14:07:40.495 +08:00] [Info] [peer.rs:2106] ["insert new region"] [region="id: 34144801 start_key: 7480000000000018FF435F7280000004AEFF9A4E430000000000FA end_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA region_epoch { conf_ver: 40837 version: 22856 } peers { id: 34144802 store_id: 11 } peers { id: 34144803 store_id: 14 } peers { id: 34144804 store_id: 1 } peers { id: 34144805 store_id: 9 }"] [region_id=34144801]
tikv_10_19_85_5_20160.log:4745:[2022/04/18 14:07:40.496 +08:00] [Info] [raft.rs:783] ["became follower at term 363"] [term=363] [raft_id=34144322] [region_id=33977192]

tikv_10_19_85_6_20162.log:3775:[2022/04/18 14:07:35.113 +08:00] [Info] [raft.rs:783] ["became follower at term 362"] [term=362] [raft_id=34144780] [region_id=33977192]
tikv_10_19_85_6_20162.log:3789:[2022/04/18 14:07:40.496 +08:00] [Info] [raft.rs:783] ["became follower at term 363"] [term=363] [raft_id=34144780] [region_id=33977192]
tikv_10_19_85_6_20162.log:3791:[2022/04/18 14:07:40.496 +08:00] [Info] [apply.rs:1234] ["execute admin command"] [command="cmd_type: BatchSplit splits { requests { split_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA new_region_id: 34144801 new_peer_ids: 34144802 new_peer_ids: 34144803 new_peer_ids: 34144804 new_peer_ids: 34144805 } right_derive: true }"] [index=4807851] [term=362] [peer_id=34144780] [region_id=33977192]
tikv_10_19_85_6_20162.log:3793:[2022/04/18 14:07:40.497 +08:00] [Info] [peer.rs:2106] ["insert new region"] [region="id: 34144801 start_key: 7480000000000018FF435F7280000004AEFF9A4E430000000000FA end_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA region_epoch { conf_ver: 40837 version: 22856 } peers { id: 34144802 store_id: 11 } peers { id: 34144803 store_id: 14 } peers { id: 34144804 store_id: 1 } peers { id: 34144805 store_id: 9 }"] [region_id=34144801]
tikv_10_19_85_6_20162.log:3909:[2022/04/18 14:07:52.541 +08:00] [Info] [raft.rs:1389] ["sends MsgTimeoutNow to 34144804 immediately as 34144804 already has up-to-date log"] [lead_transferee=34144804] [raft_id=34144805] [region_id=34144801]

tikv_10_19_85_6_20163.log:4450:[2022/04/18 14:07:40.495 +08:00] [Info] [raft.rs:1289] ["sent MsgTimeoutNow to 34144658 after received MsgAppResp"] [from=34144658] [raft_id=34144442] [region_id=33977192]
tikv_10_19_85_6_20163.log:4451:[2022/04/18 14:07:40.501 +08:00] [Info] [apply.rs:1234] ["execute admin command"] [command="cmd_type: BatchSplit splits { requests { split_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA new_region_id: 34144801 new_peer_ids: 34144802 new_peer_ids: 34144803 new_peer_ids: 34144804 new_peer_ids: 34144805 } right_derive: true }"] [index=4807851] [term=362] [peer_id=34144442] [region_id=33977192]
tikv_10_19_85_6_20163.log:4456:[2022/04/18 14:07:40.502 +08:00] [Info] [raft.rs:783] ["became follower at term 363"] [term=363] [raft_id=34144442] [region_id=33977192]
tikv_10_19_85_6_20163.log:4460:[2022/04/18 14:07:40.502 +08:00] [Info] [peer.rs:2106] ["insert new region"] [region="id: 34144801 start_key: 7480000000000018FF435F7280000004AEFF9A4E430000000000FA end_key: 7480000000000018FF435F7280000004AEFF9C78BD0000000000FA region_epoch { conf_ver: 40837 version: 22856 } peers { id: 34144802 store_id: 11 } peers { id: 34144803 store_id: 14 } peers { id: 34144804 store_id: 1 } peers { id: 34144805 store_id: 9 }"] [region_id=34144801]
```


Discussion / hints:
TiKV should also postpone split if there is an on-going transfer leader. (Suggested by @mayjiang0203, thank you!)
/assign @LykxSassinator 

Fix patch:
diff --git a/components/raftstore/src/store/fsm/peer.rs b/components/raftstore/src/store/fsm/peer.rs
index 26ca6ebe268..769877bab21 100644
--- a/components/raftstore/src/store/fsm/peer.rs
+++ b/components/raftstore/src/store/fsm/peer.rs
@@ -84,7 +84,7 @@ use crate::{
         local_metrics::{RaftMetrics, TimeTracker},
         memory::*,
         metrics::*,
-        msg::{Callback, ExtCallback, InspectedRaftMessage},
+        msg::{Callback, CampaignType, ExtCallback, InspectedRaftMessage},
         peer::{
             ConsistencyState, ForceLeaderState, Peer, PersistSnapshotResult, StaleState,
             UnsafeRecoveryExecutePlanSyncer, UnsafeRecoveryFillOutReportSyncer,
@@ -1254,8 +1254,18 @@ where
                     self.maybe_destroy();
                 }
             }
-            CasualMessage::Campaign => {
-                let _ = self.fsm.peer.raft_group.campaign();
+            CasualMessage::Campaign(campaign_type) => {
+                match campaign_type {
+                    CampaignType::ForceLeader => {
+                        // Forcely campaign to be the leader of the region.
+                        let _ = self.fsm.peer.raft_group.campaign();
+                    }
+                    CampaignType::UnsafeSplitCampaign => {
+                        // If the message is sent by the parent, it means that the parent is already
+                        // the leader of the parent region.
+                        let _ = self.fsm.peer.maybe_campaign(true);
+                    }
+                }
                 self.fsm.has_ready = true;
             }
         }
@@ -1799,7 +1809,7 @@ where
             // follower state
             let _ = self.ctx.router.send(
                 self.region_id(),
-                PeerMsg::CasualMessage(CasualMessage::Campaign),
+                PeerMsg::CasualMessage(CasualMessage::Campaign(CampaignType::ForceLeader)),
             );
         }
         self.fsm.has_ready = true;
@@ -1916,6 +1926,24 @@ where
         }
     }
 
+    #[inline]
+    /// Check whether the peer has any uncleared records in the
+    /// uncampaigned_new_regions list.
+    fn check_uncampaigned_regions(&mut self) {
+        fail_point!("on_skip_check_uncampaigned_regions", |_| {});
+        let has_uncompaigned_regions = !self
+            .fsm
+            .peer
+            .uncampaigned_new_regions
+            .as_ref()
+            .map_or(false, |r| r.is_empty());
+        // If the peer has any uncleared records in the uncampaigned_new_regions list,
+        // and there has valid leader in the region, it's safely to clear the records.
+        if has_uncompaigned_regions && self.fsm.peer.has_valid_leader() {
+            self.fsm.peer.uncampaigned_new_regions = None;
+        }
+    }
+
     fn on_raft_log_fetched(&mut self, context: GetEntriesContext, res: Box<RaftlogFetchResult>) {
         let low = res.low;
         // If the peer is not the leader anymore and it's not in entry cache warmup
@@ -2695,6 +2723,8 @@ where
 
         result?;
 
+        self.check_uncampaigned_regions();
+
         if self.fsm.peer.any_new_peer_catch_up(from_peer_id) {
             self.fsm.peer.heartbeat_pd(self.ctx);
             self.fsm.peer.should_wake_up = true;
@@ -4126,7 +4156,7 @@ where
         // It's not correct anymore, so set it to false to schedule a split check task.
         self.fsm.peer.may_skip_split_check = false;
 
-        let is_leader = self.fsm.peer.is_leader();
+        let (is_leader, is_follower) = (self.fsm.peer.is_leader(), self.fsm.peer.is_follower());
         if is_leader {
             if share_source_region_size {
                 self.fsm.peer.approximate_size = share_size;
@@ -4297,6 +4327,18 @@ where
                 .unwrap();
 
             if !campaigned {
+                // The new peer has not campaigned yet, record it for later campaign.
+                if is_follower && self.fsm.peer.region().get_peers().len() > 1 {
+                    if self.fsm.peer.uncampaigned_new_regions.is_none() {
+                        self.fsm.peer.uncampaigned_new_regions = Some(vec![]);
+                    }
+                    self.fsm
+                        .peer
+                        .uncampaigned_new_regions
+                        .as_mut()
+                        .unwrap()
+                        .push(new_region_id);
+                }
                 if let Some(msg) = meta
                     .pending_msgs
                     .swap_remove_front(|m| m.get_to_peer() == &meta_peer)
diff --git a/components/raftstore/src/store/msg.rs b/components/raftstore/src/store/msg.rs
index f6cfdec0f5f..2668984d71a 100644
--- a/components/raftstore/src/store/msg.rs
+++ b/components/raftstore/src/store/msg.rs
@@ -519,6 +519,17 @@ where
     CheckPendingAdmin(UnboundedSender<CheckAdminResponse>),
 }
 
+/// Campaign type for triggering a Raft campaign.
+#[derive(Debug, Clone, Copy)]
+pub enum CampaignType {
+    /// Forcely campaign to be the leader.
+    ForceLeader,
+    /// Campaign triggered by the leader of a parent region. It's used to make
+    /// the new splitted peer campaign to get votes.
+    /// Only if the parent region has valid leader, will it be safe to do that.
+    UnsafeSplitCampaign,
+}
+
 /// Message that will be sent to a peer.
 ///
 /// These messages are not significant and can be dropped occasionally.
@@ -611,7 +622,8 @@ pub enum CasualMessage<EK: KvEngine> {
     },
 
     // Trigger raft to campaign which is used after exiting force leader
-    Campaign,
+    // or make new splitted peers campaign to get votes.
+    Campaign(CampaignType),
 }
 
 impl<EK: KvEngine> fmt::Debug for CasualMessage<EK> {
@@ -674,7 +686,9 @@ impl<EK: KvEngine> fmt::Debug for CasualMessage<EK> {
                 "SnapshotApplied, peer_id={}, tombstone={}",
                 peer_id, tombstone
             ),
-            CasualMessage::Campaign => write!(fmt, "Campaign"),
+            CasualMessage::Campaign(_) => {
+                write!(fmt, "Campaign")
+            }
         }
     }
 }
diff --git a/components/raftstore/src/store/peer.rs b/components/raftstore/src/store/peer.rs
index a411c0f5c8f..9de2f33842d 100644
--- a/components/raftstore/src/store/peer.rs
+++ b/components/raftstore/src/store/peer.rs
@@ -74,7 +74,10 @@ use super::{
     cmd_resp,
     local_metrics::{RaftMetrics, TimeTracker},
     metrics::*,
-    peer_storage::{write_peer_state, CheckApplyingSnapStatus, HandleReadyResult, PeerStorage},
+    peer_storage::{
+        write_peer_state, CheckApplyingSnapStatus, HandleReadyResult, PeerStorage,
+        RAFT_INIT_LOG_TERM,
+    },
     read_queue::{ReadIndexQueue, ReadIndexRequest},
     transport::Transport,
     util::{
@@ -99,7 +102,7 @@ use crate::{
         },
         hibernate_state::GroupState,
         memory::{needs_evict_entry_cache, MEMTRACE_RAFT_ENTRIES},
-        msg::{CasualMessage, ErrorCallback, PeerMsg, RaftCommand, SignificantMsg, StoreMsg},
+        msg::{CampaignType, CasualMessage, ErrorCallback, RaftCommand, SignificantMsg, StoreMsg},
         peer_storage::HandleSnapshotResult,
         snapshot_backup::{AbortReason, SnapshotBrState},
         txn_ext::LocksStatus,
@@ -108,7 +111,7 @@ use crate::{
             CleanupTask, CompactTask, HeartbeatTask, RaftlogGcTask, ReadDelegate, ReadExecutor,
             ReadProgress, RegionTask, SplitCheckTask,
         },
-        Callback, Config, GlobalReplicationState, PdTask, ReadCallback, ReadIndexContext,
+        Callback, Config, GlobalReplicationState, PdTask, PeerMsg, ReadCallback, ReadIndexContext,
         ReadResponse, TxnExt, WriteCallback, RAFT_INIT_LOG_INDEX,
     },
     Error, Result,
@@ -1075,6 +1078,12 @@ where
     pub snapshot_recovery_state: Option<SnapshotBrState>,
 
     last_record_safe_point: u64,
+
+    /// Used to record uncampaigned regions, which are the new regions
+    /// created when a follower applies a split. If the follower becomes a
+    /// leader, a campaign is triggered for those regions.
+    /// Once the parent region has valid leader, this list will be cleared.
+    pub uncampaigned_new_regions: Option<Vec<u64>>,
 }
 
 impl<EK, ER> Peer<EK, ER>
@@ -1213,6 +1222,7 @@ where
             lead_transferee: raft::INVALID_ID,
             unsafe_recovery_state: None,
             snapshot_recovery_state: None,
+            uncampaigned_new_regions: None,
         };
 
         // If this region has only one peer and I am the one, campaign directly.
@@ -1754,6 +1764,11 @@ where
         self.raft_group.raft.state == StateRole::Leader
     }
 
+    #[inline]
+    pub fn is_follower(&self) -> bool {
+        self.raft_group.raft.state == StateRole::Follower && self.peer.role != PeerRole::Learner
+    }
+
     #[inline]
     pub fn is_witness(&self) -> bool {
         self.peer.is_witness
@@ -2358,12 +2373,30 @@ where
                             "region_id" => self.region_id,
                         );
                     }
+                    // After the leadership changed, send `CasualMessage::Campaign
+                    // {notify_by_parent: true}` to the target peer to campaign
+                    // leader if there exists uncampaigned regions. It's used to
+                    // ensure that a leader is elected promptly for the newly
+                    // created Raft group, minimizing availability impact (e.g.
+                    // #12410 and #17602.).
+                    if let Some(new_regions) = self.uncampaigned_new_regions.take() {
+                        for new_region in new_regions {
+                            let _ = ctx.router.send(
+                                new_region,
+                                PeerMsg::CasualMessage(CasualMessage::Campaign(
+                                    CampaignType::UnsafeSplitCampaign,
+                                )),
+                            );
+                        }
+                    }
                 }
                 StateRole::Follower => {
                     self.leader_lease.expire();
                     self.mut_store().cancel_generating_snap(None);
                     self.clear_disk_full_peers(ctx);
                     self.clear_in_memory_pessimistic_locks();
+                    // Clear the uncampaigned list.
+                    self.uncampaigned_new_regions = None;
                 }
                 _ => {}
             }
@@ -3677,6 +3710,12 @@ where
             return false;
         }
 
+        // And only if the split region does not enter election state, will it be
+        // safe to campaign.
+        if self.term() > RAFT_INIT_LOG_TERM {
+            return false;
+        }
+
         // If last peer is the leader of the region before split, it's intuitional for
         // it to become the leader of new split region.
         let _ = self.raft_group.campaign();
diff --git a/components/snap_recovery/src/leader_keeper.rs b/components/snap_recovery/src/leader_keeper.rs
index ca2623c82ca..24aee729675 100644
--- a/components/snap_recovery/src/leader_keeper.rs
+++ b/components/snap_recovery/src/leader_keeper.rs
@@ -11,7 +11,9 @@ use engine_traits::KvEngine;
 use futures::compat::Future01CompatExt;
 use raftstore::{
     errors::{Error, Result},
-    store::{Callback, CasualMessage, CasualRouter, SignificantMsg, SignificantRouter},
+    store::{
+        msg::CampaignType, Callback, CasualMessage, CasualRouter, SignificantMsg, SignificantRouter,
+    },
 };
 use tikv_util::{future::paired_future_callback, timer::GLOBAL_TIMER_HANDLE};
 
@@ -125,7 +127,7 @@ where
     }
 
     fn force_leader(&self, region_id: u64) -> Result<()> {
-        let msg = CasualMessage::Campaign;
+        let msg = CasualMessage::Campaign(CampaignType::ForceLeader);
         self.router.send(region_id, msg)?;
         // We have nothing to do...
         Ok(())
@@ -192,7 +194,7 @@ mod test {
             msg: raftstore::store::CasualMessage<EK>,
         ) -> raftstore::errors::Result<()> {
             match msg {
-                raftstore::store::CasualMessage::Campaign => {
+                raftstore::store::CasualMessage::Campaign(_) => {
                     if !self.regions.contains(&region_id) {
                         return Err(raftstore::Error::RegionNotFound(region_id));
                     }


Test patch:
diff --git a/tests/failpoints/cases/test_split_region.rs b/tests/failpoints/cases/test_split_region.rs
index aa6519c9b1c..39c6ff74366 100644
--- a/tests/failpoints/cases/test_split_region.rs
+++ b/tests/failpoints/cases/test_split_region.rs
@@ -24,7 +24,9 @@ use kvproto::{
 use pd_client::PdClient;
 use raft::eraftpb::MessageType;
 use raftstore::{
-    store::{config::Config as RaftstoreConfig, util::is_vote_msg, Callback, PeerMsg},
+    store::{
+        config::Config as RaftstoreConfig, util::is_vote_msg, Callback, PeerMsg, WriteResponse,
+    },
     Result,
 };
 use test_raftstore::*;
@@ -1405,3 +1407,139 @@ fn test_split_region_with_no_valid_split_keys() {
     rx.recv_timeout(Duration::from_secs(5)).unwrap();
     rx.try_recv().unwrap_err();
 }
+
+/// Test that if the original leader of the parent region is tranfered to
+/// another peer, the new leader of the parent region will notify the new split
+/// region to campaign.
+#[test]
+fn test_region_split_after_parent_leader_transfer() {
+    let mut cluster = test_raftstore::new_node_cluster(0, 3);
+    cluster.cfg.raft_store.raft_base_tick_interval = ReadableDuration::millis(50);
+    cluster.cfg.raft_store.raft_election_timeout_ticks = 10;
+
+    let pd_client = cluster.pd_client.clone();
+    pd_client.disable_default_operator();
+
+    cluster.run();
+    cluster.must_put(b"k1", b"v1");
+    cluster.must_put(b"k2", b"v2");
+    cluster.must_put(b"k3", b"v3");
+    // Transfer leader to peer 3.
+    let region = pd_client.get_region(b"k2").unwrap();
+    cluster.must_transfer_leader(region.get_id(), new_peer(3, 3));
+
+    // Setting: only peers on store 2 can become leader.
+    for id in 1..=3 {
+        if id == 2 {
+            continue;
+        }
+        cluster.add_send_filter(CloneFilterFactory(
+            RegionPacketFilter::new(region.get_id(), id)
+                .msg_type(MessageType::MsgRequestPreVote)
+                .direction(Direction::Send),
+        ));
+        cluster.add_send_filter(CloneFilterFactory(
+            RegionPacketFilter::new(1000, id)
+                .msg_type(MessageType::MsgRequestPreVote)
+                .direction(Direction::Send),
+        ));
+    }
+
+    // Split region to peer 1 & 2, not allow peer 3 (leader) to split.
+    let no_split_on_store_3 = "on_split";
+    fail::cfg(no_split_on_store_3, "pause").unwrap();
+    cluster.split_region(
+        &region,
+        b"k2",
+        Callback::write(Box::new(move |_write_resp: WriteResponse| {})),
+    );
+    // Wait the old lease of the leader timeout and peer 2 gets votes
+    // to become the new leader.
+    thread::sleep(
+        cluster.cfg.raft_store.raft_base_tick_interval.0
+            * cluster.cfg.raft_store.raft_election_timeout_ticks as u32
+            * 2,
+    );
+    // As the split is paused, the leader of the parent region should
+    // be peer 2, not peer 3. And peer 2 will notify the new split region
+    //  `campaign` to become leader.
+    cluster.reset_leader_of_region(region.get_id());
+    assert_eq!(
+        cluster.leader_of_region(region.get_id()).unwrap(),
+        new_peer(2, 2)
+    );
+    // The leader of the new split region should be peer 1002.
+    let new_region = pd_client.get_region(b"k1").unwrap();
+    assert_eq!(
+        cluster.leader_of_region(new_region.get_id()).unwrap(),
+        new_peer(2, 1002)
+    );
+    fail::remove(no_split_on_store_3);
+}
+
+/// Test that the leader of the new split region will not be changed after
+/// the leader of the parent region is transferred.
+#[test]
+fn test_region_split_after_new_leader_elected() {
+    let mut cluster = test_raftstore::new_node_cluster(0, 3);
+    cluster.cfg.raft_store.raft_base_tick_interval = ReadableDuration::millis(50);
+    cluster.cfg.raft_store.raft_election_timeout_ticks = 10;
+
+    let pd_client = cluster.pd_client.clone();
+    pd_client.disable_default_operator();
+
+    cluster.run();
+    cluster.must_put(b"k1", b"v1");
+    cluster.must_put(b"k2", b"v2");
+    cluster.must_put(b"k3", b"v3");
+    // Transfer leader to peer 3.
+    let region = pd_client.get_region(b"k2").unwrap();
+    cluster.must_transfer_leader(region.get_id(), new_peer(3, 3));
+
+    // Setting: only peers on store 2 can become leader.
+    for id in 1..=3 {
+        if id == 2 {
+            continue;
+        }
+        cluster.add_send_filter(CloneFilterFactory(
+            RegionPacketFilter::new(region.get_id(), id)
+                .msg_type(MessageType::MsgRequestPreVote)
+                .direction(Direction::Send),
+        ));
+    }
+
+    // Split region to peer 1 & 2, not allow peer 3 (leader) to split.
+    let skip_clear_uncampaign = "on_skip_check_uncampaigned_regions";
+    fail::cfg(skip_clear_uncampaign, "return").unwrap();
+    let no_split_on_store_3 = "on_split";
+    fail::cfg(no_split_on_store_3, "pause").unwrap();
+    cluster.split_region(
+        &region,
+        b"k2",
+        Callback::write(Box::new(move |_write_resp: WriteResponse| {})),
+    );
+    // Wait the leader of the new split region has been elected.
+    thread::sleep(
+        cluster.cfg.raft_store.raft_base_tick_interval.0
+            * cluster.cfg.raft_store.raft_election_timeout_ticks as u32
+            * 2,
+    );
+    cluster.reset_leader_of_region(region.get_id());
+    assert_eq!(
+        cluster.leader_of_region(region.get_id()).unwrap(),
+        new_peer(2, 2)
+    );
+    // The leader of the new split region should be elected.
+    let new_region = pd_client.get_region(b"k1").unwrap();
+    let new_region_leader = cluster.leader_of_region(new_region.get_id()).unwrap();
+    // The new leader will notify the new split region  `campaign` to become
+    // leader, but the leader of the new split region is already elected.
+    fail::remove(no_split_on_store_3);
+    // The leader of the new split region should not changed.
+    cluster.reset_leader_of_region(new_region.get_id());
+    assert_eq!(
+        cluster.leader_of_region(new_region.get_id()).unwrap(),
+        new_region_leader
+    );
+    fail::remove(skip_clear_uncampaign);
+}
diff --git a/tests/integrations/raftstore/test_split_region.rs b/tests/integrations/raftstore/test_split_region.rs
index 10771c57863..18762adc529 100644
--- a/tests/integrations/raftstore/test_split_region.rs
+++ b/tests/integrations/raftstore/test_split_region.rs
@@ -1278,3 +1278,43 @@ fn test_gen_split_check_bucket_ranges() {
     // the bucket_ranges should be None to refresh the bucket
     cluster.send_half_split_region_message(&region, None);
 }
+
+#[test]
+fn test_clear_uncampaigned_regions_after_split() {
+    let mut cluster = test_raftstore::new_node_cluster(0, 3);
+    cluster.cfg.raft_store.raft_base_tick_interval = ReadableDuration::millis(50);
+    cluster.cfg.raft_store.raft_election_timeout_ticks = 10;
+
+    let pd_client = cluster.pd_client.clone();
+    pd_client.disable_default_operator();
+
+    cluster.run();
+    cluster.must_put(b"k1", b"v1");
+    cluster.must_put(b"k2", b"v2");
+    cluster.must_put(b"k3", b"v3");
+    // Transfer leader to peer 3.
+    let region = pd_client.get_region(b"k2").unwrap();
+    cluster.must_transfer_leader(region.get_id(), new_peer(3, 3));
+
+    // New split regions will be recorded into uncampaigned region list of
+    // followers (in peer 1 and peer 2).
+    cluster.split_region(
+        &region,
+        b"k2",
+        Callback::write(Box::new(move |_write_resp: WriteResponse| {})),
+    );
+    // Wait the old lease of the leader timeout and followers clear its
+    // uncampaigned region list.
+    thread::sleep(
+        cluster.cfg.raft_store.raft_base_tick_interval.0
+            * cluster.cfg.raft_store.raft_election_timeout_ticks as u32
+            * 3,
+    );
+    // The leader of the parent region should still be peer 3 as no
+    // other peers can become leader.
+    cluster.reset_leader_of_region(region.get_id());
+    assert_eq!(
+        cluster.leader_of_region(region.get_id()).unwrap(),
+        new_peer(3, 3)
+    );
+}