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_async_io.rs
    path of file: ./repos/tikv/tests/failpoints/cases
    
    the code of the file until where you have to start completion: // Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.

use std::{
    sync::{atomic::AtomicBool, mpsc, Arc, Mutex},
    time::Duration,
};

use pd_client::PdClient;
use raft::eraftpb::MessageType;
use test_raftstore::*;
use test_raftstore_macro::test_case;
use tikv_util::HandyRwLock;

// Test if the entries can be committed and applied on followers even when
// leader's io is paused.
#[test_case(test_raftstore::new_node_cluster)]
#[test_case(test_raftstore_v2::new_node_cluster)]
fn test_async_io_commit_without_leader_persist() {
    let mut cluster = new_cluster(0, 3);
    cluster.cfg.raft_store.cmd_batch_concurrent_ready_max_count = 0;
    cluster.cfg.raft_store.store_io_pool_size = 2;
    let pd_client = Arc::clone(&cluster.pd_client);
    pd_client.disable_default_operator();

    cluster.run();

    let region = pd_client.get_region(b"k1").unwrap();
    let peer_1 = find_peer(&region, 1).cloned().unwrap();

    cluster.must_put(b"k1", b"v1");
    cluster.must_transfer_leader(region.get_id(), peer_1);

    let raft_before_save_on_store_1_fp = "raft_before_save_on_store_1";
    fail::cfg(raft_before_save_on_store_1_fp, "pause").unwrap();

    for i in 2..10 {
        let _ = cluster
            .async_put(format!("k{}", i).as_bytes(), b"v1")
            .unwrap();
    }

    // Although leader can not persist entries, these entries can be committed
    must_get_equal(&cluster.get_engine(2), b"k9", b"v1");
    must_get_equal(&cluster.get_engine(3), b"k9", b"v1");
    // For now, entries must be applied after persisting
    must_get_none(&cluster.get_engine(1), b"k9");

    fail::remove(raft_before_save_on_store_1_fp);
    must_get_equal(&cluster.get_engine(3),
    
fn test_async_io_cannot_handle_ready_when_persist_snapshot() {
    let mut cluster = new_node_cluster(0, 3);
    cluster.cfg.raft_store.store_io_pool_size = 2;
    configure_for_snapshot(&mut cluster.cfg);
    let pd_client = Arc::clone(&cluster.pd_client);
    pd_client.disable_default_operator();

    let r1 = cluster.run_conf_change();
    pd_client.must_add_peer(r1, new_peer(2, 2));
    pd_client.must_add_peer(r1, new_peer(3, 3));

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

    must_get_equal(&cluster.get_engine(2), b"k1", b"v1");
    must_get_equal(&cluster.get_engine(3), b"k1", b"v1");

    cluster.add_send_filter(IsolationFilterFactory::new(3));

    for i in 2..10 {
        cluster.must_put(format!("k{}", i).as_bytes(), b"v1");
    }

    must_get_equal(&cluster.get_engine(2), b"k9", b"v1");

    let raft_before_save_kv_on_store_3_fp = "raft_before_save_kv_on_store_3";
    fail::cfg(raft_before_save_kv_on_store_3_fp, "pause").unwrap();

    let (notify_tx, notify_rx) = mpsc::channel();
    cluster.sim.wl().add_recv_filter(
        3,
        Box::new(MessageTypeNotifier::new(
            MessageType::MsgSnapshot,
            notify_tx,
            Arc::new(AtomicBool::new(true)),
        )),
    );

    cluster.clear_send_filters();

    // Wait for leader sending snapshot to peer 3
    notify_rx.recv_timeout(Duration::from_secs(5)).unwrap();
    // Wait for peer 3 handling snapshot
    sleep_ms(100);

    let panic_if_handle_ready_3_fp = "panic_if_handle_ready_3";
    fail::cfg(panic_if_handle_ready_3_fp, "return").unwrap();

    for i in 10..20 {
        cluster.must_put(format!("k{}", i).as_bytes(), b"v1");
    }

    cluster.must_transfer_leader(r1, new_peer(2, 2));

    for i in 20..30 {
        cluster.must_put(format!("k{}", i).as_bytes(), b"v1");
    }

    fail::remove(panic_if_handle_ready_3_fp);
    fail::remove(raft_before_save_kv_on_store_3_fp);

    must_get_equal(&cluster.get_engine(3), b"k29", b"v1");
}