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

Code before:
    let large_val = vec![b'a'; 1024 * 1024];
    // The size of these entries should be larger than MAX_COMMITTED_SIZE_PER_READY
    for i in 0..50 {
        cluster.must_put(format!("k1{}", i).as_bytes(), large_val.as_slice());
    }
    cluster.must_put(b"k150", b"v150");

    let after_handle_catch_up_logs_for_merge_1003_fp = "after_handle_catch_up_logs_for_merge_1003";
    fail::cfg(after_handle_catch_up_logs_for_merge_1003_fp, "return").unwrap();

    fail::remove(before_handle_normal_3_fp);

    // Wait for apply state writing to kv db
    sleep_ms(200);

    cluster.shutdown();

    fail::remove(after_handle_catch_up_logs_for_merge_1003_fp);

    cluster.start().unwrap();

    must_get_equal(&cluster.get_engine(3), b"k150", b"v150");
}

// Test the consistency of EntryCache when partioned leader constains
// uncommitted propose, after the partion is recovered, it can replicate raft
// entries for new leader correctly. The case test the corner scenario that
// partitioned leader receive a new Append msg for new elected leader and the
// new entries are already committed and overlap with existing
// entries uncomitted entries in the entry cache, it may cause panic if handles
// incorrectly. See issue https://github.com/tikv/tikv/issues/17868 for more details.
#[test]
fn test_early_apply_leader_demote_by_append() {
    let mut cluster = new_node_cluster(0, 3);
    cluster.pd_client.disable_default_operator();

    cluster.run();

    cluster.must_put(b"k1", b"v1");
    let region = cluster.pd_client.get_region(b"k1").unwrap();
    cluster.must_transfer_leader(region.id, new_peer(1, 1));

    // isolate 1 for 2,3
    let dropped_append: Arc<Mutex<RaftMessage>> = Arc::new(Mutex::new(RaftMessage::default()));
    let msg_ref = dropped_append.clone();
    cluster.add_recv_filter_on_node(
        1,
        Box::new(DropMessageFilter::new(Arc::new(move |m| {
            // save dropped append.
            if m.get_message().msg_type == MessageType::MsgAppend {
                *msg_ref.lock().unwrap() = m.clone();

Review comment (suggestion):
```suggestion
// uncommitted propose, after the partition is recovered, it can replicate raft
```

Code after:
    let large_val = vec![b'a'; 1024 * 1024];
    // The size of these entries should be larger than MAX_COMMITTED_SIZE_PER_READY
    for i in 0..50 {
        cluster.must_put(format!("k1{}", i).as_bytes(), large_val.as_slice());
    }
    cluster.must_put(b"k150", b"v150");

    let after_handle_catch_up_logs_for_merge_1003_fp = "after_handle_catch_up_logs_for_merge_1003";
    fail::cfg(after_handle_catch_up_logs_for_merge_1003_fp, "return").unwrap();

    fail::remove(before_handle_normal_3_fp);

    // Wait for apply state writing to kv db
    sleep_ms(200);

    cluster.shutdown();

    fail::remove(after_handle_catch_up_logs_for_merge_1003_fp);

    cluster.start().unwrap();

    must_get_equal(&cluster.get_engine(3), b"k150", b"v150");
}

// Test the consistency of EntryCache when partitioned leader contains
// uncommitted propose, and after the partition is recovered, it can replicate
// raft entries for new leader correctly. This case tests the corner scenario
// that partitioned leader receives a new Append msg from new elected leader and
// the new entries are already committed and overlap with existing
// uncommitted entries in the entry cache, it may cause panic if handled
// incorrectly. See issue https://github.com/tikv/tikv/issues/17868 for more details.
#[test]
fn test_early_apply_leader_demote_by_append() {
    let mut cluster = new_node_cluster(0, 3);
    cluster.pd_client.disable_default_operator();

    cluster.run();

    cluster.must_put(b"k1", b"v1");
    let region = cluster.pd_client.get_region(b"k1").unwrap();
    cluster.must_transfer_leader(region.id, new_peer(1, 1));

    // isolate 1 for 2,3
    let dropped_append: Arc<Mutex<RaftMessage>> = Arc::new(Mutex::new(RaftMessage::default()));
    let msg_ref = dropped_append.clone();
    cluster.add_recv_filter_on_node(
        1,
        Box::new(DropMessageFilter::new(Arc::new(move |m| {
            // save dropped append.
            if m.get_message().msg_type == MessageType::MsgAppend {
                *msg_ref.lock().unwrap() = m.clone();