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

Code before:
    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();
            }

            false
        }))),
    );
    for id in [2, 3] {
        cluster.add_recv_filter_on_node(
            id,
            Box::new(DropMessageFilter::new(Arc::new(|m| {
                m.get_message().from != 1
            }))),
        );
    }

    // propose a new write, the write should timeout.
    let ch = cluster.async_put(b"k1", b"v2").unwrap();
    block_on_timeout(ch, Duration::from_millis(100)).unwrap_err();

    let fp = "pause_on_peer_collect_message";
    // pause peer 1 to wait for leader timeout, should new leader
    fail::cfg(fp, "pause").unwrap();

    // wait for leader timeout.
    sleep_ms(200);

    cluster.reset_leader_of_region(region.id);
    let leader = cluster.leader_of_region(region.id).unwrap();
    assert_ne!(leader.store_id, 1);

    cluster.must_put(b"k1", b"v3");

    // Send the dropped msg to store 1.
    let mut msg = dropped_append.lock().unwrap().clone();
    assert_eq!(msg.get_message().to, 1);
    // Advance the committed index of the Append msg to trigger the corner case.
    // I don't find an easy way to trigger this kind of msg, so direct modify the
    // first append to mock that scenario.
    let entry_idx = msg
        .get_message()
        .get_entries()
        .last()
        .map(|e| e.index)
        .unwrap();
    msg.mut_message().commit = entry_idx;
    let peer_msg = PeerMsg::RaftMessage(Box::new(InspectedRaftMessage { heap_size: 0, msg }), None);

Review comment (suggestion):
```suggestion
    // pause peer 1 to wait for leader timeout
```

Code after:
    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();
            }

            false
        }))),
    );
    for id in [2, 3] {
        cluster.add_recv_filter_on_node(
            id,
            Box::new(DropMessageFilter::new(Arc::new(|m| {
                m.get_message().from != 1
            }))),
        );
    }

    // propose a new write, the write should timeout.
    let ch = cluster.async_put(b"k1", b"v2").unwrap();
    block_on_timeout(ch, Duration::from_millis(100)).unwrap_err();

    let fp = "pause_on_peer_collect_message";
    // pause peer 1 to wait for leader timeout.
    fail::cfg(fp, "pause").unwrap();

    // wait for leader timeout.
    sleep_ms(200);

    cluster.reset_leader_of_region(region.id);
    let leader = cluster.leader_of_region(region.id).unwrap();
    assert_ne!(leader.store_id, 1);

    cluster.must_put(b"k1", b"v3");

    // Send the dropped msg to store 1.
    let mut msg = dropped_append.lock().unwrap().clone();
    assert_eq!(msg.get_message().to, 1);
    // Advance the committed index of the Append msg to trigger the corner case.
    // I don't find an easy way to trigger this kind of msg, so direct modify the
    // first append to mock that scenario.
    let entry_idx = msg
        .get_message()
        .get_entries()
        .last()
        .map(|e| e.index)
        .unwrap();
    msg.mut_message().commit = entry_idx;
    let peer_msg = PeerMsg::RaftMessage(Box::new(InspectedRaftMessage { heap_size: 0, msg }), None);