You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: test_early_apply.rs
    path of file: ./repos/tikv/tests/integrations/raftstore
    
    the code of the file until where you have to start completion: // Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.

use std::time::Duration;

use engine_rocks::RocksEngine;
use engine_traits::{RaftEngine, RaftEngineDebug};
use kvproto::raft_serverpb::RaftLocalState;
use raft::eraftpb::MessageType;
use raftstore::store::*;
use test_raftstore::*;

fn delete_old_data<E: RaftEngineDebug>(engine: &E, id: u64) {
    let mut deleter = engine.log_batch(0);
    let mut last_index = 0;
    engine
        .scan_entries(id, |e| {
            last_index = e.get_index();
            Ok(true)
        })
        .unwrap();
    let state = RaftLocalState {
        last_index,
        ..Default::default()
    };
    engine
        .clean(id, 0 /* first_index */, &state, &mut deleter)
        .unwrap();
    engine.consume(&mut deleter, true /* sync */).unwrap();
}

/// Allow lost situation.
#[derive(PartialEq, Clone, Copy)]
enum DataLost {
    /// The leader loses commit index.
    ///
    /// A leader can't lost both the committed entries and commit index
    /// at the same time.
    LeaderCommit,
    /// A follower loses commit index.
    FollowerCommit,
    /// All the nodes loses data.
    ///
    /// Typically, both leader and followers lose commit index.
    AllLost,
}

fn test<A, C>(
    cluster: &mut Cluster<RocksEngine, NodeCluster<RocksEngine>>,
    action: A,
    check: C,
    mode: DataLost,
) where
    A: FnOnce(&mut Cluster<RocksEngine, NodeCluster<RocksEngine>>),
    C: FnOnce(&mut Cluster<RocksEngine, NodeCluster<RocksEngine>>),
{
    let filter = match mode {
        DataLost::AllLost | DataLost::LeaderCommit => RegionPacketFilter::new(1, 1)
            .msg_type(MessageType::Msg
    
fn test_update_internal_apply_index() {
    let mut cluster = new_node_cluster(0, 4);
    cluster.pd_client.disable_default_operator();
    // So compact log will not be triggered automatically.
    configure_for_request_snapshot(&mut cluster.cfg);
    cluster.run();
    cluster.must_transfer_leader(1, new_peer(3, 3));
    cluster.must_put(b"k1", b"v1");
    must_get_equal(&cluster.get_engine(1), b"k1", b"v1");

    let filter = RegionPacketFilter::new(1, 3)
        .msg_type(MessageType::MsgAppendResponse)
        .direction(Direction::Recv);
    cluster.add_send_filter(CloneFilterFactory(filter));
    let last_index = cluster.raft_local_state(1, 1).get_last_index();
    let _ = cluster.async_remove_peer(1, new_peer(4, 4)).unwrap();
    let _ = cluster.async_put(b"k2", b"v2").unwrap();
    let mut snaps = Vec::new();
    for id in 1..3 {
        cluster.wait_last_index(1, id, last_index + 2, Duration::from_secs(3));
        snaps.push((id, cluster.get_raft_engine(id).dump_all_data(id)));
    }
    cluster.clear_send_filters();
    must_get_equal(&cluster.get_engine(1), b"k2", b"v2");
    must_get_equal(&cluster.get_engine(2), b"k2", b"v2");

    // Simulate data lost in raft cf.
    for (id, mut batch) in snaps {
        cluster.stop_node(id);
        delete_old_data(&cluster.get_raft_engine(id), id);
        cluster
            .get_raft_engine(id)
            .consume(&mut batch, true /* sync */)
            .unwrap();
        cluster.run_node(id).unwrap();
    }

    let region = cluster.get_region(b"k1");
    // Issues a heartbeat to followers so they will re-commit the logs.
    let resp = read_on_peer(
        &mut cluster,
        new_peer(3, 3),
        region,
        b"k1",
        true,
        Duration::from_secs(3),
    )
    .unwrap();
    assert!(!resp.get_header().has_error(), "{:?}", resp);
    cluster.stop_node(3);
    cluster.must_put(b"k3", b"v3");
}