Repository: tikv/tikv
File: tests/failpoints/cases/test_transfer_leader.rs (Rust)

Code before:
    pd_client.transfer_leader(region_id, new_peer(3, 3), vec![]);
    pd_client.region_leader_must_be(region_id, new_peer(3, 3));

    fail::cfg("apply_before_split_1_1", "pause").unwrap();
    fail::cfg("apply_before_split_1_3", "pause").unwrap();
    fail::cfg("propose_locks_before_transfer_leader", "return").unwrap();

    let region = pd_client.get_region(b"x1").unwrap();
    cluster.split_region(&region, "x2".as_bytes(), Callback::None);
    sleep_ms(300);
    // Expect split is pending on the current leader.
    assert_eq!(pd_client.get_regions_number(), 1);

    pd_client.transfer_leader(region_id, new_peer(2, 2), vec![]);
    sleep_ms(300);
    pd_client.region_leader_must_be(region_id, new_peer(2, 2));
    sleep_ms(300);
    // TODO(hwy): We cannot enable this assertion yet since https://github.com/tikv/tikv/issues/12410.
    // Expect split is finished on the new leader.
    // assert_eq!(pd_client.get_regions_number(), 2);
}

// This test verifies that a leader transfer is:
// - Not allowed for the source region when the transferee has been applied a
//   region commit-merge but the leader has not yet applied.
// - Allowed for the source region when the transferee has been applied a region
//   prepare-merge but the leader has not yet applied.
// - Allowed for the target region in both scenarios above.
#[test]
fn test_when_applied_region_merge_on_transferee_pessimistic_lock() {
    let mut cluster = new_server_cluster(0, 4);
    // To enable the transferee to quickly report the merged region information.
    cluster.cfg.raft_store.pd_heartbeat_tick_interval = ReadableDuration::millis(50);
    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));
    // Use peer_id 4 since we want to reuse the failpoint
    // apply_before_commit_merge_except_1_4.
    pd_client.must_add_peer(region_id, new_peer(4, 4));
    pd_client.region_leader_must_be(region_id, new_peer(1, 1));

    let region = cluster.get_region(b"x2");
    let region_id = region.id;
    pd_client.split_region(region, pdpb::CheckPolicy::Usekey, vec![b"x2".to_vec()]);
    sleep_ms(300);
    let left_region = cluster.get_region(b"x1");
    let right_region = cluster.get_region(b"x3");
    assert_eq!(region_id, right_region.get_id());
    let left_region_peer_on_store1 = new_peer(
        left_region.get_peers()[0].store_id,

Review comment (suggestion):
```suggestion
// - Not allowed for the source region when the transferee has applied a
//   region commit-merge but the leader has not yet applied.
// - Allowed for the source region when the transferee has applied a region
```

Code after:
    pd_client.region_leader_must_be(region_id, new_peer(3, 3));

    fail::cfg("apply_before_split_1_1", "pause").unwrap();
    fail::cfg("apply_before_split_1_3", "pause").unwrap();
    fail::cfg("propose_locks_before_transfer_leader", "return").unwrap();

    let region = pd_client.get_region(b"x1").unwrap();
    cluster.split_region(&region, "x2".as_bytes(), Callback::None);
    sleep_ms(300);
    // Expect split is pending on the current leader.
    assert_eq!(pd_client.get_regions_number(), 1);

    pd_client.transfer_leader(region_id, new_peer(2, 2), vec![]);
    sleep_ms(300);
    pd_client.region_leader_must_be(region_id, new_peer(2, 2));
    sleep_ms(300);
    // TODO(hwy): We cannot enable this assertion yet since https://github.com/tikv/tikv/issues/12410.
    // Expect split is finished on the new leader.
    // assert_eq!(pd_client.get_regions_number(), 2);
}

// This test verifies that a leader transfer is:
// - Not allowed for the source region when the transferee has applied a region
//   commit-merge but the leader has not yet applied.
// - Allowed for the source region when the transferee has applied a region
//   prepare-merge but the leader has not yet applied.
// - Allowed for the target region in both scenarios above.
#[test]
fn test_when_applied_region_merge_on_transferee_pessimistic_lock() {
    let mut cluster = new_server_cluster(0, 4);
    // To enable the transferee to quickly report the merged region information.
    cluster.cfg.raft_store.pd_heartbeat_tick_interval = ReadableDuration::millis(50);
    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));
    // Use peer_id 4 since we want to reuse the failpoint
    // apply_before_commit_merge_except_1_4.
    pd_client.must_add_peer(region_id, new_peer(4, 4));
    pd_client.region_leader_must_be(region_id, new_peer(1, 1));

    let region = cluster.get_region(b"x2");
    let region_id = region.id;
    pd_client.split_region(region, pdpb::CheckPolicy::Usekey, vec![b"x2".to_vec()]);
    sleep_ms(300);
    let left_region = cluster.get_region(b"x1");
    let right_region = cluster.get_region(b"x3");
    assert_eq!(region_id, right_region.get_id());
    let left_region_peer_on_store1 = new_peer(
        left_region.get_peers()[0].store_id,
        left_region.get_peers()[0].id,