raftstore: allow leader transfer if conf change applied on transferee

components/raftstore/src/store/fsm/apply.rs

@@ -2321,6 +2321,17 @@ where
         request: &AdminRequest,
     ) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)> {
         assert!(request.has_change_peer_v2());
+        fail_point!(
+            "apply_on_conf_change_1_1",
+            self.id() == 1 && self.region_id() == 1,
+            |_| unreachable!()
+        );
+        fail_point!(
+            "apply_on_conf_change_3_1",
+            self.id() == 3 && self.region_id() == 1,
+            |_| unreachable!()
+        );
+
         let changes = request.get_change_peer_v2().get_change_peers().to_vec();
 
         info!(

components/raftstore/src/store/fsm/peer.rs

@@ -3689,7 +3689,19 @@ where
         }
     }
 
+    // NOTE: This method is used by both the leader and the follower.
+    // Both the request and response for transfer-leader share the MessageType
+    // `MsgTransferLeader`.
     fn on_transfer_leader_msg(&mut self, msg: &eraftpb::Message, peer_disk_usage: DiskUsage) {
+        info!(
+            "received transferring leader";
+            "region_id" => self.fsm.region_id(),
+            "peer_id" => self.fsm.peer.peer_id(),
+            "from" => msg.get_from(),
+            "term" => self.fsm.peer.term(),
+            "msg_term" => msg.get_log_term(),
+        );
+
         // log_term is set by original leader, represents the term last log is written
         // in, which should be equal to the original leader's term.
         if msg.get_log_term() != self.fsm.peer.term() {

components/raftstore/src/store/peer.rs

@@ -58,7 +58,7 @@ use tikv_util::{
     box_err,
     codec::number::decode_u64,
     debug, error, info,
-    store::find_peer_by_id,
+    store::{find_peer_by_id, is_learner},
     sys::disk::DiskUsage,
     time::{duration_to_sec, monotonic_raw_now, Instant as TiInstant, InstantExt},
     warn,
@@ -3921,17 +3921,6 @@ where
         extra_msgs: Vec<ExtraMessage>,
         ctx: &mut PollContext<EK, ER, T>,
     ) -> bool {
-        // Checks if safe to transfer leader.
-        if self.raft_group.raft.has_pending_conf() {
-            info!(
-                "reject transfer leader due to pending conf change";
-                "region_id" => self.region_id,
-                "peer_id" => self.peer.get_id(),
-                "peer" => ?peer,
-            );
-            return false;
-        }
-
         // Broadcast heartbeat to make sure followers commit the entries immediately.
         // It's only necessary to ping the target peer, but ping all for simplicity.
         self.raft_group.ping();
@@ -3987,9 +3976,19 @@ where
             }
         }
 
-        if self.raft_group.raft.has_pending_conf()
-            || self.raft_group.raft.pending_conf_index > index
-        {
+        // It's safe to transfer leader to a target peer that has already applied the
+        // configuration change, even if the current leader has not yet applied
+        // it. For more details, refer to the issue at:
+        // https://github.com/tikv/tikv/issues/17363#issuecomment-2404227253.
+        if self.raft_group.raft.pending_conf_index > index {
+            info!(
+                "not ready to transfer leader; target peer has an unapplied conf change";
+                "region_id" => self.region_id,
+                "target_peer_id" => peer_id,
+                "pending_conf_index" => self.raft_group.raft.pending_conf_index,
+                "leader_applied_index" => self.raft_group.raft.raft_log.applied,
+                "target_applied_index" => index
+            );
             return Some("pending conf change");
         }
 
@@ -4686,7 +4685,7 @@ where
     ) -> bool {
         let pending_snapshot = self.is_handling_snapshot() || self.has_pending_snapshot();
         // shouldn't transfer leader to witness peer or non-witness waiting data
-        if self.is_witness() || self.wait_data
+        if self.is_witness() || is_learner(&self.peer) || self.wait_data
             || pending_snapshot
             || msg.get_from() != self.leader_id()
             // Transfer leader to node with disk full will lead to write availablity downback.
@@ -4703,6 +4702,7 @@ where
                 "pending_snapshot" => pending_snapshot,
                 "disk_usage" => ?ctx.self_disk_usage,
                 "is_witness" => self.is_witness(),
+                "is_learner" => is_learner(&self.peer),
                 "wait_data" => self.wait_data,
             );
             return true;
@@ -4771,6 +4771,14 @@ where
         &mut self,
         reply_cmd: bool, // whether it is a reply to a TransferLeader command
     ) {
+        info!(
+            "ack transfer leader";
+            "region_id" => self.region_id,
+            "from_peer" => self.peer_id(),
+            "to_peer" => self.leader_id(),
+            "reply_cmd" => reply_cmd,
+        );
+
         let mut msg = eraftpb::Message::new();
         msg.set_from(self.peer_id());
         msg.set_to(self.leader_id());

components/test_pd_client/src/pd.rs

@@ -1383,15 +1383,21 @@ impl TestPdClient {
     pub fn region_leader_must_be(&self, region_id: u64, peer: metapb::Peer) {
         for _ in 0..500 {
             sleep_ms(10);
-            if let Some(p) = self.cluster.rl().leaders.get(&region_id) {
-                if *p == peer {
-                    return;
-                }
+            if self.check_region_leader(region_id, peer.clone()) {
+                return;
             }
         }
         panic!("region {} must have leader: {:?}", region_id, peer);
     }
 
+    pub fn check_region_leader(&self, region_id: u64, peer: metapb::Peer) -> bool {
+        self.cluster
+            .rl()
+            .leaders
+            .get(&region_id)
+            .map_or(false, |p| *p == peer)
+    }
+
     // check whether region is split by split_key or not.
     pub fn check_split(&self, region: &metapb::Region, split_key: &[u8]) -> bool {
         // E.g, 1 [a, c) -> 1 [a, b) + 2 [b, c)

tests/failpoints/cases/test_conf_change.rs

@@ -12,7 +12,8 @@ use pd_client::PdClient;
 use raft::eraftpb::{ConfChangeType, MessageType};
 use test_raftstore::*;
 use test_raftstore_macro::test_case;
-use tikv_util::{config::ReadableDuration, HandyRwLock};
+use test_util::init_log_for_test;
+use tikv_util::{config::ReadableDuration, logger::init_log, HandyRwLock};
 
 #[test_case(test_raftstore::new_node_cluster)]
 #[test_case(test_raftstore_v2::new_node_cluster)]
@@ -316,3 +317,93 @@ fn test_handle_conf_change_when_apply_fsm_resume_pending_state() {
         cluster.must_put(format!("kk{}", i).as_bytes(), b"v1");
     }
 }
+
+// This test simulates a scenario where a configuration change has been applied
+// on the target peer, allowing a leader transfer to that target peer even if
+// the change hasn't been applied on the current leader.
+//
+// The setup involves a 4-node cluster where peer-1 starts as the leader. A
+// configuration change is initiated to remove peer-2. This configuration
+// change commits successfully but only applies on peer-4. Meanwhile, it
+// remains pending on the leader (peer-1) and on peer-3.
+//
+// The expected result is that under these circumstances, a leader transfer
+// cannot occur to peer-3 due to its unapplied configuration change. However,
+// a transfer to peer-4 should succeed since it has already applied the
+// change.
+#[test]
+fn test_applied_conf_change_on_target_peer_allows_transfer_leader() {
+    let mut cluster = new_server_cluster(0, 4);
+    let pd_client = cluster.pd_client.clone();
+    pd_client.disable_default_operator();
+    let region_id = cluster.run_conf_change();
+    pd_client.must_add_peer(region_id, new_peer(2, 2));
+    pd_client.must_add_peer(region_id, new_peer(3, 3));
+    pd_client.must_add_peer(region_id, new_peer(4, 4));
+
+    cluster.must_put(b"k1", b"v1");
+
+    fail::cfg("apply_on_conf_change_1_1", "pause").unwrap();
+    fail::cfg("apply_on_conf_change_3_1", "pause").unwrap();
+
+    pd_client.remove_peer(region_id, new_peer(2, 2));
+    // Wait for remove_peer.
+    sleep_ms(300);
+    // Peer 2 is still exists since the leader hasn't applied the ConfChange
+    // yet.
+    pd_client.must_have_peer(region_id, new_peer(2, 2));
+
+    // Use async_put for insertion here to avoid timeout errors, as synchronize put
+    // would hang due to the leader's apply process being paused.
+    cluster.async_put(b"k2", b"v2");
+
+    pd_client.transfer_leader(region_id, new_peer(3, 3), vec![]);
+    // Wait for transfer_leader.
+    sleep_ms(300);
+    assert_eq!(
+        pd_client.check_region_leader(region_id, new_peer(3, 3)),
+        false
+    );
+
+    pd_client.transfer_leader(region_id, new_peer(4, 4), vec![]);
+    pd_client.region_leader_must_be(region_id, new_peer(4, 4));
+
+    // Verify the data completeness on the new leader.
+    must_get_equal(&cluster.get_engine(4), b"k1", b"v1");
+    must_get_equal(&cluster.get_engine(4), b"k2", b"v2");
+}
+
+// This test verifies that a leader transfer is rejected when the target peer
+// has been demoted to a learner but the leader has not yet applied this
+// configuration change.
+#[test]
+fn test_applied_conf_change_on_target_learner_rejects_transfer_leader() {
+    let mut cluster = new_server_cluster(0, 3);
+    let pd_client = cluster.pd_client.clone();
+    pd_client.disable_default_operator();
+    let region_id = cluster.run_conf_change();
+    pd_client.must_add_peer(region_id, new_peer(2, 2));
+    pd_client.must_add_peer(region_id, new_peer(3, 3));
+    pd_client.region_leader_must_be(region_id, new_peer(1, 1));
+
+    fail::cfg("apply_on_conf_change_1_1", "pause").unwrap();
+
+    // Demote peer-2 to be a learner.
+    cluster.pd_client.joint_confchange(
+        region_id,
+        vec![(ConfChangeType::AddLearnerNode, new_learner_peer(2, 2))],
+    );
+    // Wait for joint_confchange.
+    sleep_ms(300);
+
+    pd_client.transfer_leader(region_id, new_peer(2, 2), vec![]);
+    // Wait for transfer_leader.
+    sleep_ms(300);
+    assert_eq!(
+        pd_client.check_region_leader(region_id, new_peer(2, 2)),
+        false
+    );
+
+    pd_client.transfer_leader(region_id, new_peer(3, 3), vec![]);
+    pd_client.region_leader_must_be(region_id, new_peer(3, 3));
+}