// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2022 PingCAP, Inc.

use std::{
    sync::*,
    time::{Duration, Instant},
};

use api_version::{KvFormat, test_kv_format_impl};
use cdc::{Task, Validate};
use concurrency_manager::ConcurrencyManager;
use futures::{SinkExt, executor::block_on};
use grpcio::WriteFlags;
use kvproto::{cdcpb::*, kvrpcpb::*, metapb::PeerRole};
use pd_client::PdClient;
use raft::eraftpb::MessageType;
use test_pd_client::PdClientExt;
use test_raftstore::{
    must_check_txn_status, must_kv_commit, new_peer, sleep_ms, try_kv_prewrite_pessimistic,
};
use test_rfstore::*;
use tikv_util::config::ReadableDuration;
use txn_types::{Key, Lock, LockType, TimeStamp};

use crate::{
    CloudTestSuiteBuilder, TestSuite, configure_for_lease_read_cfg, get_ready_region,
    get_stable_region_and_leader, new_event_feed, new_event_feed_v2, subscribe_until_initialized,
};

fn assert_same_peer_identity(actual: &kvproto::metapb::Peer, expected: &kvproto::metapb::Peer) {
    assert_eq!(actual.get_id(), expected.get_id());
    assert_eq!(actual.get_store_id(), expected.get_store_id());
}

fn is_same_peer_identity(actual: &kvproto::metapb::Peer, expected: &kvproto::metapb::Peer) -> bool {
    actual.get_id() == expected.get_id() && actual.get_store_id() == expected.get_store_id()
}

fn must_get_stable_transfer_leader_target(
    suite: &mut TestSuite,
    key: &[u8],
    region_id: u64,
    leader: &kvproto::metapb::Peer,
) -> kvproto::metapb::Peer {
    let timer = Instant::now();
    loop {
        let (region, current_leader) = get_stable_region_and_leader(suite, key);
        if region.get_id() == region_id && is_same_peer_identity(&current_leader, leader) {
            if let Some(peer) = region
                .get_peers()
                .iter()
                .find(|p| {
                    p.get_role() == PeerRole::Voter && !is_same_peer_identity(p, &current_leader)
                })
                .cloned()
            {
                return peer;
            }
        }

        assert!(
            timer.elapsed() <= Duration::from_secs(15),
            "timed out waiting for region {} to expose a stable voter transferee while leader {:?} is still serving",
            region_id,
            leader
        );
        sleep_ms(100);
    }
}

#[test]
fn test_cdc_basic() {
    let mut suite = TestSuite::new(1, ApiVersion::V2);

    let region = get_ready_region(&mut suite, b"xkey0");
    let region_id = region.get_id();
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_basic initial subscribe",
    );

    // Sleep a while to make sure the stream is registered.
    sleep_ms(1000);
    // There must be a delegate.
    let scheduler = suite.endpoints.values().next().unwrap().clone();
    scheduler
        .schedule(Task::Validate(Validate::Region(
            region_id,
            Box::new(|delegate| {
                let d = delegate.unwrap();
                assert_eq!(d.downstreams().len(), 1);
            }),
        )))
        .unwrap();

    // If tikv enable ApiV2, txn key needs to start with 'x';
    let (k, v) = ("xkey1".to_owned(), "value".to_owned());
    // Prewrite
    let start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = k.clone().into_bytes();
    mutation.value = v.into_bytes();
    suite.must_kv_prewrite(region_id, vec![mutation], k.clone().into_bytes(), start_ts);

    let mut counter = 0;
    loop {
        // Even if there is no write,
        // resolved ts should be advanced regularly.
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            assert_ne!(0, resolved_ts.ts);
            counter += 1;
        }
        if counter > 5 {
            break;
        }
    }
    // Commit
    let commit_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k.into_bytes()], start_ts, commit_ts);
    let mut event = receive_event(false);
    let mut events = event.take_events();
    assert_eq!(events.len(), 1, "{:?}", event);
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Entries(entries) => {
            assert_eq!(entries.entries.len(), 1);
            assert_eq!(entries.entries[0].get_type(), EventLogType::Committed);
        }
        other => panic!("unknown event {:?}", other),
    }

    // Split region 1
    let region1 = suite.cluster.get_region(&[]);
    suite.cluster.must_split(&region1, b"xkey2");
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    let split_regions = match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Error(err) => {
            assert!(err.has_epoch_not_match(), "{:?}", err);
            err.get_epoch_not_match().get_current_regions().to_vec()
        }
        other => panic!("unknown event {:?}", other),
    };
    // The delegate must be removed.
    scheduler
        .schedule(Task::Validate(Validate::Region(
            region_id,
            Box::new(|delegate| {
                assert!(delegate.is_none());
            }),
        )))
        .unwrap();

    // request again.
    // Cloud-engine keeps the original region id on the right half after split,
    // so the left half containing `xkey0` is the newly created region.
    // Reuse the split response instead of querying the region cache again:
    // immediately after split, the cache can still report the previous epoch and
    // CDC will correctly reject the resubscribe request as stale.
    let left_region = split_regions
        .iter()
        .find(|region| region.get_id() != region_id)
        .unwrap();
    let region_id2 = left_region.get_id();
    suite.cluster.wait_region_info_ready(left_region);
    // The new left-half region may not be queryable through `get_context()` yet,
    // so build the request directly from the split response instead of going back
    // through `new_changedata_request()`.
    let mut req = ChangeDataRequest {
        region_id: region_id2,
        ..Default::default()
    };
    req.mut_header()
        .set_cluster_id(suite.cluster.pd_client().get_cluster_id().unwrap());
    req.set_region_epoch(left_region.get_region_epoch().clone());
    let (mut req_tx, resp_rx) = suite
        .get_region_cdc_client(region_id2)
        .event_feed()
        .unwrap();
    event_feed_wrap.replace(Some(resp_rx));
    let timer = Instant::now();
    let mut events = loop {
        block_on(req_tx.send((req.clone(), WriteFlags::default()))).unwrap();
        let events = receive_event(false).events.to_vec();
        assert_eq!(events.len(), 1);
        match events[0].event.as_ref() {
            Some(Event_oneof_event::Error(err)) if err.has_region_not_found() => {
                assert!(
                    timer.elapsed() <= Duration::from_secs(5),
                    "timed out waiting for split region {} to become subscribable: {:?}",
                    region_id2,
                    err
                );
                sleep_ms(20);
            }
            _ => break events,
        }
    };
    assert_eq!(events.len(), 1);
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Entries(es) => {
            // The re-subscribed downstream now follows the left split region,
            // which already contains the committed `xkey1` write. Cloud-engine
            // replays that scanned history before sending `Initialized`.
            assert_eq!(es.entries.len(), 2, "{:?}", es);
            let committed = &es.entries[0];
            assert_eq!(committed.get_type(), EventLogType::Committed, "{:?}", es);
            assert_eq!(committed.key, b"xkey1", "{:?}", es);

            let initialized = &es.entries[1];
            assert_eq!(
                initialized.get_type(),
                EventLogType::Initialized,
                "{:?}",
                es
            );
        }
        other => panic!("unknown event {:?}", other),
    }
    // Sleep a while to make sure the stream is registered.
    sleep_ms(200);
    scheduler
        .schedule(Task::Validate(Validate::Region(
            region_id2,
            Box::new(|delegate| {
                let d = delegate.unwrap();
                assert_eq!(d.downstreams().len(), 1);
            }),
        )))
        .unwrap();

    // Drop both halves of the bidi stream. Keeping `req_tx` alive leaves the
    // client request stream open, so the server keeps the downstream
    // registered and `Validate::Region` still sees the delegate.
    drop(req_tx);
    event_feed_wrap.replace(None);
    // Sleep a while to make sure the stream is deregistered.
    sleep_ms(200);
    scheduler
        .schedule(Task::Validate(Validate::Region(
            region_id2,
            Box::new(|delegate| {
                assert!(delegate.is_none());
            }),
        )))
        .unwrap();

    // Stale region epoch.
    let mut req = suite.new_changedata_request(region_id2);
    req.set_region_epoch(Default::default()); // Zero region epoch.
    let (mut req_tx, resp_rx) = suite
        .get_region_cdc_client(region_id2)
        .event_feed()
        .unwrap();
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    event_feed_wrap.replace(Some(resp_rx));
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Error(err) => {
            assert!(err.has_epoch_not_match(), "{:?}", err);
        }
        other => panic!("unknown event {:?}", other),
    }

    suite.stop();
}

#[test]
fn test_cdc_not_leader() {
    test_kv_format_impl!(test_cdc_not_leader_impl<ApiV2>);
}

fn test_cdc_not_leader_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(3, F::TAG);

    // `TestSuite::new()` pre-splits the magic keyspace ranges during bootstrap.
    // Wait until the chosen region and its leader stop changing, otherwise the
    // initial subscription can fail with bootstrap-time `NotLeader` before this
    // test performs its own explicit leader transfer.
    let (region, leader) = get_stable_region_and_leader(&mut suite, b"xkey1");
    let region_id = region.get_id();
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_not_leader initial subscribe",
    );
    // Sleep a while to make sure the stream is registered.
    sleep_ms(1000);
    // There must be a delegate.
    let scheduler = suite.endpoints.get(&leader.get_store_id()).unwrap().clone();
    let (tx, rx) = mpsc::channel();
    let tx_ = tx.clone();
    scheduler
        .schedule(Task::Validate(Validate::Region(
            region_id,
            Box::new(move |delegate| {
                let d = delegate.unwrap();
                assert_eq!(d.downstreams().len(), 1);
                tx_.send(()).unwrap();
            }),
        )))
        .unwrap();
    rx.recv_timeout(Duration::from_secs(1)).unwrap();
    assert!(
        suite
            .obs
            .get(&leader.get_store_id())
            .unwrap()
            .is_subscribed(region_id)
            .is_some()
    );

    // Bootstrap keeps rebalancing replicas for a short time after the region
    // layout becomes readable. Wait until this exact region still has the same
    // leader and also exposes a non-leader voter before issuing transfer leader.
    let peer = must_get_stable_transfer_leader_target(&mut suite, b"xkey1", region_id, &leader);
    suite.cluster.must_transfer_leader(region_id, peer.clone());
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Error(err) => {
            // When the old leader loses leadership during CDC initialization,
            // rfstore can return `stale command` before a `NotLeader` header is
            // built. CDC intentionally maps such request errors to
            // `region_not_found`, so accept both representations here.
            if err.has_not_leader() {
                // The old leader may step down before it knows who won the new
                // term, in which case CDC forwards a `NotLeader` error without
                // filling the leader field.
                if err.get_not_leader().has_leader() {
                    // CDC only needs to tell clients which peer to reconnect to.
                    // The protobuf may omit or normalize auxiliary fields such as
                    // `role`, so compare leader identity instead of the entire
                    // peer message.
                    assert_same_peer_identity(err.get_not_leader().get_leader(), &peer);
                }
            } else {
                assert!(err.has_region_not_found(), "{:?}", err);
            }
        }
        other => panic!("unknown event {:?}", other),
    }
    assert!(
        suite
            .obs
            .get(&leader.get_store_id())
            .unwrap()
            .is_subscribed(region_id)
            .is_none()
    );

    // Sleep a while to make sure the stream is deregistered.
    sleep_ms(200);
    scheduler
        .schedule(Task::Validate(Validate::Region(
            region_id,
            Box::new(move |delegate| {
                assert!(delegate.is_none());
                tx.send(()).unwrap();
            }),
        )))
        .unwrap();
    rx.recv_timeout(Duration::from_millis(200)).unwrap();

    // PD may report the transferee as leader before the region is fully ready to
    // serve CDC initialization on that node. Wait until the same region/leader
    // stays stable before opening a new stream to the new leader.
    let (region_after_transfer, leader_after_transfer) =
        get_stable_region_and_leader(&mut suite, b"xkey1");
    assert_eq!(region_after_transfer.get_id(), region_id);
    assert_same_peer_identity(&leader_after_transfer, &peer);

    // Reconnect through the current leader instead of reusing the stale stream on
    // the old leader. The old leader path above already verified that it reports
    // a leadership-change error and cleans up its delegate after leadership
    // changes.
    let (mut req_tx, resp_rx) = suite.get_region_cdc_client(region_id).event_feed().unwrap();
    event_feed_wrap.replace(Some(resp_rx));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_not_leader reconnect subscribe",
    );

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_cdc_cluster_id_mismatch() {
    test_kv_format_impl!(test_cdc_cluster_id_mismatch_impl<ApiV2>);
}

fn test_cdc_cluster_id_mismatch_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(3, F::TAG);

    let region = get_ready_region(&mut suite, b"");
    let region_id = region.id;
    // Send request with mismatched cluster id.
    let mut req = suite.new_changedata_request(region_id);
    req.mut_header().set_ticdc_version("5.3.0".into());
    req.mut_header().cluster_id += 1;
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    block_on(req_tx.send((req.clone(), WriteFlags::default()))).unwrap();

    // Assert mismatch.
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Error(err) => {
            assert!(err.has_cluster_id_mismatch(), "{:?}", err);
        }
        other => panic!("unknown event {:?}", other),
    }

    // Low version request.
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        req.mut_header().set_ticdc_version("4.0.8".into());
        req.mut_header().set_cluster_id(DEFAULT_CLUSTER_ID + 1);
        let (mut req_tx, event_feed_wrap, receive_event) =
            new_event_feed(suite.get_region_cdc_client(region_id));
        subscribe_until_initialized(
            &mut req_tx,
            receive_event.as_ref(),
            &req,
            "test_cdc_cluster_id_mismatch low-version subscribe",
        );

        event_feed_wrap.replace(None);
    }

    suite.stop();
}

#[test]
fn test_cdc_stale_epoch_after_region_ready() {
    test_kv_format_impl!(test_cdc_stale_epoch_after_region_ready_impl<ApiV2>);
}

fn test_cdc_stale_epoch_after_region_ready_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(3, F::TAG);

    // Like `test_cdc_not_leader`, avoid the left-most pre-split region. This
    // case wants a region that is already ready so a zero epoch is rejected as
    // stale, not one that is still finishing leader election after bootstrap.
    let region = get_ready_region(&mut suite, b"xkey1");
    let region_id = region.get_id();
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_stale_epoch_after_region_ready initial subscribe",
    );

    let mut req = suite.new_changedata_request(region_id);
    req.set_region_epoch(Default::default()); // zero epoch is always stale.
    let (mut req_tx, resp_rx) = suite.get_region_cdc_client(region_id).event_feed().unwrap();
    let _resp_rx = event_feed_wrap.replace(Some(resp_rx));
    block_on(req_tx.send((req.clone(), WriteFlags::default()))).unwrap();
    // Must receive epoch not match error.
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Error(err) => {
            assert!(err.has_epoch_not_match(), "{:?}", err);
        }
        other => panic!("unknown event {:?}", other),
    }

    // The zero-epoch request only proves CDC can reject stale epochs. Bootstrap
    // may still be finishing replica changes and can move leadership after that
    // response is returned. Reconnect through the current stable leader before
    // retrying with the real epoch, otherwise the old stream can race into a
    // transient `NotLeader`.
    let stable_region = get_ready_region(&mut suite, b"xkey1");
    assert_eq!(stable_region.get_id(), region_id);
    req.set_region_epoch(suite.get_context(region_id).take_region_epoch());
    let (mut req_tx, resp_rx) = suite.get_region_cdc_client(region_id).event_feed().unwrap();
    let _resp_rx = event_feed_wrap.replace(Some(resp_rx));
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    // Must receive epoch not match error.
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    match events.pop().unwrap().event.unwrap() {
        // Even if there is no write,
        // it should always outputs an Initialized event.
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 1, "{:?}", es);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Initialized, "{:?}", es);
        }
        Event_oneof_event::Error(err) => {
            assert!(err.has_epoch_not_match(), "{:?}", err);
        }
        other => panic!("unknown event {:?}", other),
    }

    // Cancel event feed before finishing test.
    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_cdc_scan() {
    test_kv_format_impl!(test_cdc_scan_impl<ApiV2>);
}

fn test_cdc_scan_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    // `presplit_magic_keyspace()` updates PD first and stores apply the split
    // asynchronously. Query the cluster view instead of taking a one-shot PD
    // lookup so the test starts from the region that has already become usable.
    let region = get_ready_region(&mut suite, b"xkey1");
    let region_id = region.id;

    let (k, v) = (b"xkey1".to_vec(), b"value".to_vec());
    // Prewrite
    let start_ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = k.clone();
    mutation.value = v.clone();
    suite.must_kv_prewrite(region_id, vec![mutation], k.clone(), start_ts1);
    // Commit
    let commit_ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k.clone()], start_ts1, commit_ts1);

    // Prewrite again
    let start_ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = k.clone();
    mutation.value = v.clone();
    suite.must_kv_prewrite(region_id, vec![mutation], k.clone(), start_ts2);

    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    let mut events = receive_event(false).events.to_vec();
    if events.len() == 1 {
        events.extend(receive_event(false).events.into_iter());
    }
    assert_eq!(events.len(), 2, "{:?}", events);
    match events.remove(0).event.unwrap() {
        // Batch size is set to 2.
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 1, "{:?}", es);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Committed, "{:?}", es);
            assert_eq!(e.start_ts, start_ts1.into_inner(), "{:?}", es);
            assert_eq!(e.commit_ts, commit_ts1.into_inner(), "{:?}", es);
            assert_eq!(e.key, k, "{:?}", es);
            assert_eq!(e.value, v, "{:?}", es);
        }
        other => panic!("unknown event {:?}", other),
    }
    match events.pop().unwrap().event.unwrap() {
        // Then it outputs Initialized event.
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 1, "{:?}", es);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Initialized, "{:?}", es);
        }
        other => panic!("unknown event {:?}", other),
    }

    // checkpoint_ts = 6;
    let checkpoint_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    // Commit = 7;
    let commit_ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k.clone()], start_ts2, commit_ts2);
    // Prewrite delete
    // Start = 8;
    let start_ts3 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Del);
    mutation.key = k.clone();
    suite.must_kv_prewrite(region_id, vec![mutation], k.clone(), start_ts3);

    let mut req = suite.new_changedata_request(region_id);
    req.checkpoint_ts = checkpoint_ts.into_inner();
    let (mut req_tx, resp_rx) = suite.get_region_cdc_client(region_id).event_feed().unwrap();
    event_feed_wrap.replace(Some(resp_rx));
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    let mut events = receive_event(false).events.to_vec();
    if events.len() == 1 {
        events.extend(receive_event(false).events.to_vec());
    }
    assert_eq!(events.len(), 2, "{:?}", events);
    match events.remove(0).event.unwrap() {
        // Batch size is set to 2.
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 1, "{:?}", es);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Committed, "{:?}", es);
            assert_eq!(e.get_op_type(), EventRowOpType::Put, "{:?}", es);
            assert_eq!(e.start_ts, start_ts2.into_inner(), "{:?}", es);
            assert_eq!(e.commit_ts, commit_ts2.into_inner(), "{:?}", es);
            assert_eq!(e.key, k, "{:?}", es);
            assert_eq!(e.value, v, "{:?}", es);
        }
        other => panic!("unknown event {:?}", other),
    }
    assert_eq!(events.len(), 1, "{:?}", events);
    match events.pop().unwrap().event.unwrap() {
        // Then it outputs Initialized event.
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 1, "{:?}", es);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Initialized, "{:?}", es);
        }
        other => panic!("unknown event {:?}", other),
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_cdc_tso_failure() {
    test_kv_format_impl!(test_cdc_tso_failure_impl<ApiV2>);
}

fn test_cdc_tso_failure_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(3, F::TAG);

    // The left-most bootstrap region is still settling after the magic keyspace
    // split. Pick a stable `xkey` region so this case only validates resolved-ts
    // progress under TSO failures.
    let region = get_ready_region(&mut suite, b"xkey1");
    let region_id = region.get_id();
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_tso_failure initial subscribe",
    );

    suite.cluster.pd_client().trigger_tso_failure();

    // Make sure resolved ts can be advanced normally even with few tso failures.
    let mut counter = 0;
    let mut previous_ts = 0;
    loop {
        // Even if there is no write,
        // resolved ts should be advanced regularly.
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            assert!(resolved_ts.ts >= previous_ts);
            assert_eq!(resolved_ts.regions, vec![region_id]);
            previous_ts = resolved_ts.ts;
            counter += 1;
        }
        if counter > 5 {
            break;
        }
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_region_split() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(1)
        .after_cluster_bootstrapped(|server| {
            server.cluster.get_pd_client().disable_default_operator()
        })
        .cfg_fun(|_id, cfg| cfg.cdc.hibernate_regions_compatible = false)
        .build();

    let region = suite.cluster.get_region(&[]);
    let mut req = suite.new_changedata_request(region.get_id());
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region.get_id()));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_region_split initial subscribe",
    );
    // Split region.
    suite.cluster.must_split(&region, b"k0");
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    let split_regions = match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Error(err) => {
            assert!(err.has_epoch_not_match(), "{:?}", err);
            err.get_epoch_not_match().get_current_regions().to_vec()
        }
        other => panic!("unknown event {:?}", other),
    };
    // Try to subscribe region again.
    // Reuse the authoritative region metadata returned by EpochNotMatch.
    // Immediately querying `cluster.get_region()` after `must_split()` is racy on
    // cloud-engine: the local region cache can still report the pre-split epoch,
    // and CDC will correctly reject the resubscribe request as stale.
    let region = split_regions
        .iter()
        .find(|region| region.get_id() == req.get_region_id())
        .unwrap();
    assert_eq!(req.get_region_id(), region.get_id());
    // `EpochNotMatch` tells us the correct post-split epoch immediately, but
    // the region may still be absent from the leader's RegionInfoAccessor for a
    // brief window. Wait until CDC can resolve the region again before
    // re-subscribing, otherwise the request can race into `region_not_found`.
    suite.cluster.wait_region_info_ready(region);
    let mut subscribe_until_initialized = |req: ChangeDataRequest, context: &str| {
        subscribe_until_initialized(&mut req_tx, receive_event.as_ref(), &req, context);
    };

    req.set_region_epoch(region.get_region_epoch().clone());
    subscribe_until_initialized(req.clone(), "after split");

    // Try to subscribe region again.
    let region1 = split_regions
        .iter()
        .find(|region| region.get_id() != req.get_region_id())
        .unwrap();
    suite.cluster.wait_region_info_ready(region1);
    let mut req1 = req;
    req1.region_id = region1.get_id();
    req1.set_region_epoch(region1.get_region_epoch().clone());
    subscribe_until_initialized(req1, "after split 1");

    // Make sure resolved ts can be advanced normally.
    let mut counter = 0;
    let mut previous_ts = 0;
    loop {
        // Even if there is no write,
        // resolved ts should be advanced regularly.
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            assert!(resolved_ts.ts >= previous_ts);
            assert!(
                resolved_ts.regions == vec![region.id, region1.id]
                    || resolved_ts.regions == vec![region1.id, region.id],
                "rts.regions = {:?}, expected = {:?}",
                resolved_ts.regions,
                vec![region.id, region1.id]
            );
            previous_ts = resolved_ts.ts;
            counter += 1;
        }
        if counter > 5 {
            break;
        }
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_duplicate_subscribe() {
    test_kv_format_impl!(test_duplicate_subscribe_impl<ApiV2>);
}

fn test_duplicate_subscribe_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(3, F::TAG);

    // This test only cares about duplicate registration semantics, so avoid the
    // bootstrap region that may still be electing right after the magic keyspace
    // split and would turn the first subscribe into an unrelated `NotLeader`.
    let region = get_ready_region(&mut suite, b"xkey1");
    let region_id = region.get_id();
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    suite.cluster.wait_region_info_ready(&region);
    subscribe_until_initialized(&mut req_tx, receive_event.as_ref(), &req, "");
    // Try to subscribe again.
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1);
    // Should receive duplicate request error.
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Error(err) => {
            assert!(err.has_duplicate_request(), "{:?}", err);
        }
        other => panic!("unknown event {:?}", other),
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_cdc_batch_size_limit() {
    test_kv_format_impl!(test_cdc_batch_size_limit_impl<ApiV2>);
}

fn test_cdc_batch_size_limit_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);

    let region = get_ready_region(&mut suite, b"xk");
    let region_id = region.id;
    // Prewrite
    let start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut m1 = Mutation::default();
    let k1 = b"xk1".to_vec();
    m1.set_op(Op::Put);
    m1.key = k1.clone();
    m1.value = vec![0; 6 * 1024 * 1024];
    let mut m2 = Mutation::default();
    let k2 = b"xk2".to_vec();
    m2.set_op(Op::Put);
    m2.key = k2.clone();
    m2.value = b"v2".to_vec();
    suite.must_kv_prewrite(region_id, vec![m1, m2], k1.clone(), start_ts);
    // Commit
    let commit_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1, k2], start_ts, commit_ts);

    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    let mut events = receive_event(false).events.to_vec();
    assert_eq!(events.len(), 1, "{:?}", events.len());
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 1);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Committed, "{:?}", e.get_type());
            assert_eq!(e.key, b"xk1", "{:?}", e.key);
        }
        other => panic!("unknown event {:?}", other),
    }
    // For the rest 2 events, Committed and Initialized.
    let mut entries = vec![];
    while entries.len() < 2 {
        match receive_event(false).events.remove(0).event.unwrap() {
            Event_oneof_event::Entries(es) => {
                entries.extend(es.entries.into_iter());
            }
            other => panic!("unknown event {:?}", other),
        }
    }
    assert_eq!(entries.len(), 2, "{:?}", entries);
    let e = &entries[0];
    assert_eq!(e.get_type(), EventLogType::Committed, "{:?}", e.get_type());
    assert_eq!(e.key, b"xk2", "{:?}", e.key);
    let e = &entries[1];
    assert_eq!(
        e.get_type(),
        EventLogType::Initialized,
        "{:?}",
        e.get_type()
    );

    // Prewrite
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        let start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
        let mut m3 = Mutation::default();
        let k3 = b"xk3".to_vec();
        m3.set_op(Op::Put);
        m3.key = k3.clone();
        m3.value = vec![0; 7 * 1024 * 1024];
        let mut m4 = Mutation::default();
        let k4 = b"xk4".to_vec();
        m4.set_op(Op::Put);
        m4.key = k4;
        m4.value = b"v4".to_vec();
        suite.must_kv_prewrite(region_id, vec![m3, m4], k3, start_ts);

        let mut events = receive_event(false).events.to_vec();
        assert_eq!(events.len(), 1, "{:?}", events);
        match events.pop().unwrap().event.unwrap() {
            Event_oneof_event::Entries(mut es) => {
                let mut entries = es.take_entries().into_vec();
                assert_eq!(entries.len(), 2);
                entries.sort_by(|a, b| a.key.cmp(&b.key));

                let e = &entries[0];
                assert_eq!(e.get_type(), EventLogType::Prewrite, "{:?}", e.get_type());
                assert_eq!(e.key, b"xk3", "{:?}", e.key);
                let e = &entries[1];
                assert_eq!(e.get_type(), EventLogType::Prewrite, "{:?}", e.get_type());
                assert_eq!(e.key, b"xk4", "{:?}", e.key);
            }
            other => panic!("unknown event {:?}", other),
        }
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_old_value_basic() {
    test_kv_format_impl!(test_old_value_basic_impl<ApiV2>);
}

fn test_old_value_basic_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"xk");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_old_value_basic initial subscribe",
    );

    // Insert value
    let mut m1 = Mutation::default();
    let k1 = b"xk1".to_vec();
    m1.set_op(Op::Insert);
    m1.key = k1.clone();
    m1.value = b"v1".to_vec();
    let ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m1], k1.clone(), ts1);
    let ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1.clone()], ts1, ts2);
    // Rollback
    let mut m2 = Mutation::default();
    m2.set_op(Op::Put);
    m2.key = k1.clone();
    m2.value = b"v2".to_vec();
    let ts3 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m2], k1.clone(), ts3);
    suite.must_kv_rollback(region_id, vec![k1.clone()], ts3);
    // Update value
    let mut m3 = Mutation::default();
    m3.set_op(Op::Put);
    m3.key = k1.clone();
    m3.value = vec![b'3'; 5120];
    let ts4 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m3], k1.clone(), ts4);
    let ts5 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1.clone()], ts4, ts5);
    // Lock
    let mut m4 = Mutation::default();
    m4.set_op(Op::Lock);
    m4.key = k1.clone();
    let ts6 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m4], k1.clone(), ts6);
    let ts7 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1.clone()], ts6, ts7);
    // Delete value and rollback
    let mut m5 = Mutation::default();
    m5.set_op(Op::Del);
    m5.key = k1.clone();
    let ts8 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m5], k1.clone(), ts8);
    suite.must_kv_rollback(region_id, vec![k1.clone()], ts8);
    // Update value
    let mut m6 = Mutation::default();
    m6.set_op(Op::Put);
    m6.key = k1.clone();
    m6.value = b"v6".to_vec();
    let ts9 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let ts10 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m6], k1.clone(), ts10);
    let ts11 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1.clone()], ts10, ts11);
    // Delete value in pessimistic txn.
    // In pessimistic txn, CDC must use for_update_ts to read the old value.
    let mut m7 = Mutation::default();
    m7.set_op(Op::PessimisticLock);
    m7.key = k1.clone();
    let ts12 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_acquire_pessimistic_lock(region_id, vec![m7.clone()], k1.clone(), ts9, ts12);
    m7.set_op(Op::Del);
    suite.must_kv_pessimistic_prewrite(region_id, vec![m7], k1.clone(), ts9, ts12);
    let ts13 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1.clone()], ts9, ts13);
    // Insert value again
    let mut m8 = Mutation::default();
    m8.set_op(Op::Insert);
    m8.key = k1.clone();
    m8.value = b"v1".to_vec();
    let ts14 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m8], k1, ts14);

    let mut event_count = 0;
    loop {
        let events = receive_event(false).events.to_vec();
        for event in events.into_iter() {
            match event.event.unwrap() {
                Event_oneof_event::Entries(mut es) => {
                    for row in es.take_entries().to_vec() {
                        if row.get_type() == EventLogType::Committed {
                            if row.get_start_ts() == ts3.into_inner()
                                || row.get_start_ts() == ts4.into_inner()
                            {
                                assert_eq!(row.get_old_value(), b"v1", "{:?}", row);
                                event_count += 1;
                            } else if row.get_start_ts() == ts8.into_inner() {
                                assert_eq!(
                                    row.get_old_value(),
                                    vec![b'3'; 5120].as_slice(),
                                    "{:?}",
                                    row
                                );
                                event_count += 1;
                            } else if row.get_start_ts() == ts9.into_inner() {
                                assert_eq!(row.get_old_value(), b"v6", "{:?}", row);
                                event_count += 1;
                            }
                        }
                    }
                }
                other => panic!("unknown event {:?}", other),
            }
        }
        // Prewrite are filtered out. (origin 4)
        if event_count >= 2 {
            break;
        }
    }

    let (mut req_tx, resp_rx) = suite.get_region_cdc_client(region_id).event_feed().unwrap();
    event_feed_wrap.replace(Some(resp_rx));
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    let mut event_count = 0;
    loop {
        let event = receive_event(false);
        for e in event.events.into_iter() {
            match e.event.unwrap() {
                Event_oneof_event::Entries(mut es) => {
                    for row in es.take_entries().to_vec() {
                        if row.get_type() == EventLogType::Committed
                            && row.get_start_ts() == ts1.into_inner()
                        {
                            assert_eq!(row.get_old_value(), b"");
                            event_count += 1;
                        } else if row.get_type() == EventLogType::Committed
                            && row.get_start_ts() == ts4.into_inner()
                        {
                            assert_eq!(row.get_old_value(), b"v1");
                            event_count += 1;
                        } else if row.get_type() == EventLogType::Prewrite
                            && row.get_start_ts() == ts14.into_inner()
                        {
                            assert_eq!(row.get_old_value(), b"");
                            event_count += 1;
                        }
                    }
                }
                other => panic!("unknown event {:?}", other),
            }
        }
        // Prewrite are filtered out. (origin 3)
        if event_count >= 2 {
            break;
        }
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_old_value_multi_changefeeds() {
    test_kv_format_impl!(test_old_value_multi_changefeeds_impl<ApiV2>);
}

fn test_old_value_multi_changefeeds_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"xk");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    let (mut req_tx_1, event_feed_wrap_1, receive_event_1) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx_1,
        receive_event_1.as_ref(),
        &req,
        "test_old_value_multi_changefeeds read-old-value subscribe",
    );

    req.set_extra_op(ExtraOp::Noop);
    let (mut req_tx_2, event_feed_wrap_2, receive_event_2) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx_2,
        receive_event_2.as_ref(),
        &req,
        "test_old_value_multi_changefeeds noop subscribe",
    );

    // Insert value
    let mut m1 = Mutation::default();
    let k1 = b"xk1".to_vec();
    m1.set_op(Op::Put);
    m1.key = k1.clone();
    m1.value = b"v1".to_vec();
    let ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m1], k1.clone(), ts1);
    let ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1.clone()], ts1, ts2);

    // Update value
    let mut m2 = Mutation::default();
    m2.set_op(Op::Put);
    m2.key = k1.clone();
    m2.value = vec![b'3'; 5120];
    let ts3 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prewrite(region_id, vec![m2], k1.clone(), ts3);
    let ts4 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![k1], ts3, ts4);

    // The downstream 1 can get old values as expected.
    let mut event_count = 0;
    loop {
        let events = receive_event_1(false).events.to_vec();
        for event in events.into_iter() {
            match event.event.unwrap() {
                Event_oneof_event::Entries(mut es) => {
                    for row in es.take_entries().to_vec() {
                        if row.get_type() == EventLogType::Committed {
                            if row.get_start_ts() == ts3.into_inner() {
                                assert_eq!(row.get_old_value(), b"v1");
                            } else {
                                assert_eq!(row.get_old_value(), b"");
                            }
                            event_count += 1;
                        }
                    }
                }
                other => panic!("unknown event {:?}", other),
            }
        }
        if event_count >= 2 {
            break;
        }
    }

    // The downstream 2 can also get old values because `req`.`extra_op` field is
    // ignored now.
    event_count = 0;
    loop {
        let events = receive_event_2(false).events.to_vec();
        for event in events.into_iter() {
            match event.event.unwrap() {
                Event_oneof_event::Entries(mut es) => {
                    for row in es.take_entries().to_vec() {
                        if row.get_type() == EventLogType::Committed {
                            if row.get_start_ts() == ts3.into_inner() {
                                assert_eq!(row.get_old_value(), b"v1");
                            } else {
                                assert_eq!(row.get_old_value(), b"");
                            }
                            event_count += 1;
                        }
                    }
                }
                other => panic!("unknown event {:?}", other),
            }
        }
        if event_count >= 2 {
            break;
        }
    }

    event_feed_wrap_1.replace(None);
    event_feed_wrap_2.replace(None);
    suite.stop();
}

#[test]
#[ignore = "Test Cloud Engine uses actual time to allocate timestamp. Cannot reset to `99`."]
fn test_cdc_resolve_ts_checking_concurrency_manager() {
    test_kv_format_impl!(test_cdc_resolve_ts_checking_concurrency_manager_impl<ApiV2>);
}

fn test_cdc_resolve_ts_checking_concurrency_manager_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"");
    let region_id = region.id;
    let leader = suite.cluster.leader_of_region(region_id).unwrap();
    let store_id = leader.get_store_id();
    let cm: ConcurrencyManager = suite.get_txn_concurrency_manager(store_id).unwrap();
    let lock_key = |key: &[u8], ts: u64| {
        let guard = block_on(cm.lock_key(&Key::from_raw(key)));
        guard.with_lock(|l| {
            *l = Some(Lock::new(
                LockType::Put,
                key.to_vec(),
                ts.into(),
                0,
                None,
                0.into(),
                1,
                ts.into(),
            ))
        });
        guard
    };

    cm.update_max_ts(20.into(), "").unwrap();

    let guard = lock_key(b"a", 80);
    suite.set_tso(99);

    let mut req = suite.new_changedata_request(region_id);
    req.set_checkpoint_ts(100);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_resolve_ts_checking_concurrency_manager initial subscribe",
    );

    fn check_resolved_ts(event: ChangeDataEvent, check_fn: impl Fn(u64)) {
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            check_fn(resolved_ts.ts)
        }
    }

    check_resolved_ts(receive_event(true), |ts| assert_eq!(ts, 80));
    assert!(cm.max_ts() >= 100.into());

    drop(guard);
    for retry in 0.. {
        let event = receive_event(true);
        let mut current_rts = 0;
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            current_rts = resolved_ts.ts;
            if resolved_ts.ts >= 100 {
                break;
            }
        }
        if retry >= 5 {
            panic!(
                "resolved ts didn't push properly after unlocking memlock. current resolved_ts: {}",
                current_rts
            );
        }
    }

    let _guard = lock_key(b"xa", 90);
    // The resolved_ts should be blocked by the mem lock but it's already greater
    // than 90. Retry until receiving an unchanged resolved_ts because the first
    // several resolved ts received might be updated before acquiring the lock.
    let mut last_resolved_ts = 0;
    let mut success = false;
    for _ in 0..5 {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            let ts = resolved_ts.ts;
            assert!(ts > 100);
            if ts == last_resolved_ts {
                success = true;
                break;
            }
            assert!(ts > last_resolved_ts);
            last_resolved_ts = ts;
        }
    }
    assert!(success, "resolved_ts not blocked by the memory lock");

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_cdc_1pc() {
    test_kv_format_impl!(test_cdc_1pc_impl<ApiV2>);
}

fn test_cdc_1pc_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"xkey0");
    let region_id = region.id;
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_1pc initial subscribe",
    );

    let (k1, v1) = (b"xkeyxk1", b"v1");
    let (k2, v2) = (b"xkeyxk2", &[0u8; 512]);

    let start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();

    // Let resolved_ts update.
    sleep_ms(500);

    // Prewrite
    let mut prewrite_req = PrewriteRequest::default();
    prewrite_req.set_context(suite.get_context(region_id));
    let mut m1 = Mutation::default();
    m1.set_op(Op::Put);
    m1.key = k1.to_vec();
    m1.value = v1.to_vec();
    prewrite_req.mut_mutations().push(m1);
    let mut m2 = Mutation::default();
    m2.set_op(Op::Put);
    m2.key = k2.to_vec();
    m2.value = v2.to_vec();
    prewrite_req.mut_mutations().push(m2);
    prewrite_req.primary_lock = k1.to_vec();
    prewrite_req.start_version = start_ts.into_inner();
    prewrite_req.lock_ttl = prewrite_req.start_version + 1;
    prewrite_req.set_try_one_pc(true);
    let prewrite_resp = suite
        .get_tikv_client(region_id)
        .kv_prewrite(&prewrite_req)
        .unwrap();
    assert!(prewrite_resp.get_one_pc_commit_ts() > 0);

    let mut resolved_ts = 0;
    loop {
        let mut cde = receive_event(true);
        if cde.get_resolved_ts().get_ts() > resolved_ts {
            resolved_ts = cde.get_resolved_ts().get_ts();
        }
        let events = cde.mut_events();
        if !events.is_empty() {
            assert_eq!(events.len(), 1);
            match events.pop().unwrap().event.unwrap() {
                Event_oneof_event::Entries(mut es) => {
                    let mut entries = es.take_entries().into_vec();
                    assert_eq!(entries.len(), 2);
                    entries.sort_by(|a, b| a.key.cmp(&b.key));

                    let (e0, e1) = (&entries[0], &entries[1]);
                    assert_eq!(e0.get_type(), EventLogType::Committed);
                    assert_eq!(e0.get_key(), k1);
                    assert_eq!(e0.get_value(), v1);
                    assert!(e0.commit_ts > resolved_ts);
                    assert_eq!(e1.get_type(), EventLogType::Committed);
                    assert_eq!(e1.get_key(), k2);
                    assert_eq!(e1.get_value(), v2);
                    assert!(e1.commit_ts > resolved_ts);
                    break;
                }
                other => panic!("unknown event {:?}", other),
            }
        }
    }

    suite.stop();
}

#[test]
fn test_old_value_1pc() {
    test_kv_format_impl!(test_old_value_1pc_impl<ApiV2>);
}

fn test_old_value_1pc_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"xk");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        &receive_event,
        &req,
        "test_old_value_1pc initial subscribe",
    );

    // Insert value
    let mut m1 = Mutation::default();
    let k1 = b"xk1".to_vec();
    m1.set_op(Op::Put);
    m1.key = k1.clone();
    m1.value = b"v1".to_vec();
    suite.must_kv_prewrite(region_id, vec![m1], k1.clone(), 10.into());
    suite.must_kv_commit(region_id, vec![k1.clone()], 10.into(), 15.into());

    // Prewrite with 1PC
    let start_ts = 20;
    let mut prewrite_req = PrewriteRequest::default();
    prewrite_req.set_context(suite.get_context(region_id));
    let mut m2 = Mutation::default();
    m2.set_op(Op::Put);
    m2.key = k1.clone();
    m2.value = b"v2".to_vec();
    prewrite_req.mut_mutations().push(m2);
    prewrite_req.primary_lock = k1;
    prewrite_req.start_version = start_ts;
    prewrite_req.lock_ttl = 1000;
    prewrite_req.set_try_one_pc(true);
    let prewrite_resp = suite
        .get_tikv_client(region_id)
        .kv_prewrite(&prewrite_req)
        .unwrap();
    assert!(prewrite_resp.get_one_pc_commit_ts() > 0);

    'outer: loop {
        let events = receive_event(false).events.to_vec();
        for event in events.into_iter() {
            match event.event.unwrap() {
                Event_oneof_event::Entries(mut es) => {
                    for row in es.take_entries().to_vec() {
                        if row.get_type() == EventLogType::Committed
                            && row.get_start_ts() == start_ts
                        {
                            assert_eq!(row.get_old_value(), b"v1");
                            break 'outer;
                        }
                    }
                }
                other => panic!("unknown event {:?}", other),
            }
        }
    }

    suite.stop();
}

#[test]
#[ignore = "Still need to fix."]
fn test_region_created_replicate() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(2)
        .after_cluster_bootstrapped(|c| c.cluster.get_pd_client().disable_default_operator())
        .build();

    let region = get_ready_region(&mut suite, &[]);
    let leader = suite.cluster.leader_of_region(region.id).unwrap();
    let transfee = region
        .peers
        .iter()
        .find(|p| p.id != leader.id)
        .cloned()
        .unwrap();
    suite.cluster.must_transfer_leader(region.id, transfee);
    suite
        .cluster
        .pd_client()
        .must_remove_peer(region.id, leader);

    let recv_filter = Box::new(
        RegionPacketFilter::new(region.get_id(), 1)
            .direction(Direction::Recv)
            .msg_type(MessageType::MsgAppend),
    );
    suite.cluster.add_cloud_recv_filter(1, recv_filter);
    suite
        .cluster
        .pd_client()
        .must_add_peer(region.id, new_peer(1, 1));
    let region = get_ready_region(&mut suite, &[]);
    let req = suite.new_changedata_request(region.id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region.id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_region_created_replicate initial subscribe",
    );
    sleep_ms(1000);
    suite.cluster.clear_recv_filters(1);

    let mut counter = 0;
    let mut previous_ts = 0;
    loop {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            assert!(resolved_ts.ts >= previous_ts);
            assert!(resolved_ts.regions == vec![region.id]);
            previous_ts = resolved_ts.ts;
            counter += 1;
        }
        if counter > 5 {
            break;
        }
    }
    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_cdc_scan_ignore_gc_fence() {
    test_kv_format_impl!(test_cdc_scan_ignore_gc_fence_impl<ApiV2>);
}

fn test_cdc_scan_ignore_gc_fence_impl<F: KvFormat>() {
    // This case is similar to `test_cdc_scan` but constructs a case with GC Fence.
    let mut suite = TestSuite::new(1, F::TAG);

    let region = get_ready_region(&mut suite, b"xkey1");
    let region_id = region.id;
    let (key, v1, v2) = (b"xkey1", b"value1", b"value2");

    // Write two versions to the key.
    let start_ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = key.to_vec();
    mutation.value = v1.to_vec();
    suite.must_kv_prewrite(region_id, vec![mutation], key.to_vec(), start_ts1);

    let commit_ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![key.to_vec()], start_ts1, commit_ts1);

    let start_ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mutation = Mutation {
        key: key.to_vec(),
        value: v2.to_vec(),
        ..Default::default()
    };
    suite.must_kv_prewrite(region_id, vec![mutation], key.to_vec(), start_ts2);

    let commit_ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![key.to_vec()], start_ts2, commit_ts2);

    // Assume the first version above is written by async commit and it's commit_ts
    // is not unique. Use it's commit_ts as another transaction's start_ts.
    // Run check_txn_status on commit_ts1 so that gc_fence will be set on the first
    // version.
    let caller_start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let action = suite.must_check_txn_status(
        region_id,
        key.to_vec(),
        commit_ts1,
        caller_start_ts,
        caller_start_ts,
        true,
    );
    assert_eq!(action, Action::LockNotExistRollback);

    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();
    let mut events = receive_event(false).events.to_vec();
    if events.len() == 1 {
        events.extend(receive_event(false).events.into_iter());
    }
    assert_eq!(events.len(), 2, "{:?}", events);
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 2, "{:?}", es);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Committed, "{:?}", es);
            assert_eq!(e.start_ts, start_ts2.into_inner(), "{:?}", es);
            assert_eq!(e.commit_ts, commit_ts2.into_inner(), "{:?}", es);
            assert_eq!(e.key, key.to_vec(), "{:?}", es);
            assert_eq!(e.value, v2.to_vec(), "{:?}", es);
            let e = &es.entries[1];
            assert_eq!(e.get_type(), EventLogType::Committed, "{:?}", es);
            assert_eq!(e.start_ts, start_ts1.into_inner(), "{:?}", es);
            assert_eq!(e.commit_ts, commit_ts1.into_inner(), "{:?}", es);
            assert_eq!(e.key, key.to_vec(), "{:?}", es);
            assert_eq!(e.value, v1.to_vec(), "{:?}", es);
        }
        other => panic!("unknown event {:?}", other),
    }
    match events.pop().unwrap().event.unwrap() {
        Event_oneof_event::Entries(es) => {
            assert!(es.entries.len() == 1, "{:?}", es);
            let e = &es.entries[0];
            assert_eq!(e.get_type(), EventLogType::Initialized, "{:?}", es);
        }
        other => panic!("unknown event {:?}", other),
    }

    suite.stop();
}

#[test]
fn test_cdc_extract_rollback_if_gc_fence_set() {
    test_kv_format_impl!(test_cdc_extract_rollback_if_gc_fence_set_impl<ApiV2>);
}

fn test_cdc_extract_rollback_if_gc_fence_set_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);

    let region = get_ready_region(&mut suite, b"xkey");
    let region_id = region.id;
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_extract_rollback_if_gc_fence_set initial subscribe",
    );

    sleep_ms(1000);

    // Write two versions of a key
    let (key, v1, v2, v3) = (b"xkey", b"value1", b"value2", b"value3");
    let start_ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = key.to_vec();
    mutation.value = v1.to_vec();
    suite.must_kv_prewrite(region_id, vec![mutation], key.to_vec(), start_ts1);

    let commit_ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![key.to_vec()], start_ts1, commit_ts1);

    let start_ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = key.to_vec();
    mutation.value = v2.to_vec();
    suite.must_kv_prewrite(region_id, vec![mutation], key.to_vec(), start_ts2);

    let commit_ts2 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_commit(region_id, vec![key.to_vec()], start_ts2, commit_ts2);

    // We don't care about the events caused by the previous writings in this test
    // case, and it's too complicated to check them. Just skip them here, and
    // wait for resolved_ts to be pushed to a greater value than the two
    // versions' commit_ts-es.
    let skip_to_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    loop {
        let e = receive_event(true);
        if let Some(r) = e.resolved_ts.as_ref() {
            if r.ts > skip_to_ts.into_inner() {
                break;
            }
        }
    }

    // Assume the two versions of the key are written by async commit transactions,
    // and their commit_ts-es are also other transaction's start_ts-es. Run
    // check_txn_status on the commit_ts-es of the two versions to cause
    // overlapping rollback.
    let caller_start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_check_txn_status(
        region_id,
        key.to_vec(),
        commit_ts1,
        caller_start_ts,
        caller_start_ts,
        true,
    );

    // Expects receiving rollback
    let event = receive_event(false);
    event
        .events
        .into_iter()
        .for_each(|e| match e.event.unwrap() {
            Event_oneof_event::Entries(es) => {
                assert!(es.entries.len() == 1, "{:?}", es);
                let e = &es.entries[0];
                assert_eq!(e.get_type(), EventLogType::Rollback, "{:?}", es);
                assert_eq!(e.get_start_ts(), commit_ts1.into_inner());
                assert_eq!(e.get_commit_ts(), 0);
            }
            other => panic!("unknown event {:?}", other),
        });

    suite.must_check_txn_status(
        region_id,
        key.to_vec(),
        commit_ts2,
        caller_start_ts,
        caller_start_ts,
        true,
    );

    // Expects receiving rollback
    let event = receive_event(false);
    event
        .events
        .into_iter()
        .for_each(|e| match e.event.unwrap() {
            Event_oneof_event::Entries(es) => {
                assert!(es.entries.len() == 1, "{:?}", es);
                let e = &es.entries[0];
                assert_eq!(e.get_type(), EventLogType::Rollback, "{:?}", es);
                assert_eq!(e.get_start_ts(), commit_ts2.into_inner());
                assert_eq!(e.get_commit_ts(), 0);
            }
            other => panic!("unknown event {:?}", other),
        });

    // In some special cases, a newly committed record may carry an overlapped
    // rollback initially. In this case, gc_fence shouldn't be set, and CDC
    // ignores the rollback and handles the committing normally.
    let start_ts3 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = key.to_vec();
    mutation.value = v3.to_vec();
    suite.must_kv_prewrite(region_id, vec![mutation], key.to_vec(), start_ts3);

    // Again, assume the transaction is committed with async commit protocol, and
    // the commit_ts is also another transaction's start_ts.
    let commit_ts3 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    // Rollback another transaction before committing, then the rolling back
    // information will be recorded in the lock.
    let caller_start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_check_txn_status(
        region_id,
        key.to_vec(),
        commit_ts3,
        caller_start_ts,
        caller_start_ts,
        true,
    );
    // Expects receiving rollback
    let event = receive_event(false);
    event
        .events
        .into_iter()
        .for_each(|e| match e.event.unwrap() {
            Event_oneof_event::Entries(es) => {
                assert!(es.entries.len() == 1, "{:?}", es);
                let e = &es.entries[0];
                assert_eq!(e.get_type(), EventLogType::Rollback, "{:?}", es);
                assert_eq!(e.get_start_ts(), commit_ts3.into_inner());
                assert_eq!(e.get_commit_ts(), 0);
            }
            other => panic!("unknown event {:?}", other),
        });
    // Commit the transaction, then it will have overlapped rollback initially.
    suite.must_kv_commit(region_id, vec![key.to_vec()], start_ts3, commit_ts3);
    // Expects receiving a normal committing event.
    let event = receive_event(false);
    event
        .events
        .into_iter()
        .for_each(|e| match e.event.unwrap() {
            Event_oneof_event::Entries(es) => {
                assert!(es.entries.len() == 1, "{:?}", es);
                let e = &es.entries[0];
                assert_eq!(e.get_type(), EventLogType::Committed, "{:?}", es);
                assert_eq!(e.get_start_ts(), start_ts3.into_inner());
                assert_eq!(e.get_commit_ts(), commit_ts3.into_inner());
                assert_eq!(e.get_value(), v3);
            }
            other => panic!("unknown event {:?}", other),
        });

    suite.stop();
}

// This test is created for covering the case that term was increased without
// leader change. Ideally leader id and term in StoreMeta should be updated
// together with a yielded SoftState, but sometimes the leader was transferred
// to another store and then changed back, a follower would not get a new
// SoftState.
#[test]
#[ignore = "packet filtering wasn't supported yet (TODO!)"]
fn test_term_change() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(3)
        .after_cluster_bootstrapped(|c| c.cluster.wait_region_replicated(b"", 3))
        .build();

    let region = get_ready_region(&mut suite, &[]);
    suite.cluster.pd_client().disable_default_operator();
    let mut peers = region.get_peers().to_vec();
    let leader = suite.cluster.leader_of_region(region.id).unwrap();
    let leader_idx = peers.iter().position(|p| p.id == leader.id).unwrap();
    peers.swap(0, leader_idx);
    let transfee = peers[1].clone();

    suite.cluster.must_transfer_leader(region.id, transfee);
    // Simulate network partition.
    let recv_filter = Box::new(
        RegionPacketFilter::new(region.get_id(), leader.store_id).direction(Direction::Recv),
    );
    suite
        .cluster
        .add_cloud_recv_filter(leader.store_id, recv_filter);
    // Transfer leader to peer 3 and then change it back to peer 2.
    // Peer 1 would not get a new SoftState.
    suite
        .cluster
        .must_transfer_leader(region.id, peers[2].clone());
    suite
        .cluster
        .must_transfer_leader(region.id, peers[1].clone());
    suite.cluster.clear_recv_filters(leader.store_id);

    suite
        .cluster
        .pd_client()
        .must_remove_peer(region.id, peers[2].clone());
    let region = get_ready_region(&mut suite, &[]);
    let req = suite.new_changedata_request(region.id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region.id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_term_change initial subscribe",
    );
    let mut counter = 0;
    let mut previous_ts = 0;
    loop {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            assert!(resolved_ts.ts >= previous_ts);
            assert!(resolved_ts.regions == vec![region.id]);
            previous_ts = resolved_ts.ts;
            counter += 1;
        }
        if counter > 5 {
            break;
        }
    }
    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_cdc_no_write_corresponding_to_lock() {
    test_kv_format_impl!(test_cdc_no_write_corresponding_to_lock_impl<ApiV2>);
}

fn test_cdc_no_write_corresponding_to_lock_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"xkey");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_no_write_corresponding_to_lock initial subscribe",
    );

    // Txn1 commit_ts = 15
    let mut m1 = Mutation::default();
    let k1 = b"xkey1".to_vec();
    m1.set_op(Op::Put);
    m1.key = k1.clone();
    m1.value = b"v1".to_vec();
    suite.must_kv_prewrite(region_id, vec![m1.clone()], k1.clone(), 10.into());
    suite.must_kv_commit(region_id, vec![k1.clone()], 10.into(), 15.into());

    // Txn2 start_ts = 15
    m1.value = b"v2".to_vec();
    suite.must_kv_prewrite(region_id, vec![m1.clone()], k1.clone(), 15.into());
    // unprotected rollback, no write is written
    suite.must_kv_rollback(region_id, vec![k1.clone()], 15.into());

    // Write a new txn
    m1.value = b"v3".to_vec();
    suite.must_kv_prewrite(region_id, vec![m1], k1.clone(), 20.into());
    suite.must_kv_commit(region_id, vec![k1], 20.into(), 25.into());

    let mut advance_cnt = 0;
    loop {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            advance_cnt += 1;
            if resolved_ts.ts >= 25 {
                break;
            }
            if advance_cnt > 50 {
                panic!("resolved_ts is not advanced, stuck at {}", resolved_ts.ts);
            }
        }
    }

    suite.stop();
}

#[test]
fn test_cdc_write_rollback_when_no_lock() {
    test_kv_format_impl!(test_cdc_write_rollback_when_no_lock_impl<ApiV2>);
}

fn test_cdc_write_rollback_when_no_lock_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"xk");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_write_rollback_when_no_lock initial subscribe",
    );

    let start_ts = block_on(suite.cluster.pd_client().get_tso())
        .unwrap()
        .into_inner();

    let mut m1 = Mutation::default();
    let k1 = b"xk1".to_vec();
    m1.set_op(Op::Put);
    m1.key = k1.clone();
    m1.value = vec![b'x'; 16];
    suite.must_kv_prewrite(region_id, vec![m1], k1.clone(), start_ts.into());

    // Wait until resolved_ts is blocked by the prewrite lock.
    loop {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            if resolved_ts.ts == start_ts {
                break;
            }
        }
    }

    // Do a rollback on the same key, but the start_ts is different.
    suite.must_kv_rollback(region_id, vec![k1.clone()], (start_ts - 5).into());

    // resolved_ts shouldn't advance beyond the unresolved prewrite lock.
    for _ in 0..10 {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            if resolved_ts.ts > start_ts {
                panic!("resolved_ts shouldn't advance beyond the prewrite lock");
            }
        }
    }

    let commit_ts = start_ts + 5;
    suite.must_kv_commit(region_id, vec![k1], start_ts.into(), commit_ts.into());

    let mut advance_cnt = 0;
    loop {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            advance_cnt += 1;
            if resolved_ts.ts > commit_ts {
                break;
            }
            if advance_cnt > 10 {
                panic!("resolved_ts is not advanced, stuck at {}", resolved_ts.ts);
            }
        }
    }

    suite.stop();
}

#[test]
#[cfg(NO_NEXT_GEN_COMPATIBLE)]
#[ignore = "TiKV cannot upgrade to Cloud Engine, and Cloud Engine doesn't support ADVANCE_METHOD = 1 yet."]
fn test_resolved_ts_cluster_upgrading() {
    let cluster = new_server_cluster(0, 3);
    cluster.pd_client.disable_default_operator();
    unsafe {
        cluster
            .pd_client
            .feature_gate()
            .reset_version("4.0.0")
            .unwrap();
    }
    let mut suite = build_testsuite_from_cluster(cluster);

    let region = get_ready_region(&mut suite, &[]);
    let req = suite.new_changedata_request(region.id);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region.id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_resolved_ts_cluster_upgrading initial subscribe",
    );
    let event = receive_event(true);
    if let Some(resolved_ts) = event.resolved_ts.as_ref() {
        assert!(resolved_ts.regions == vec![region.id]);
        assert_eq!(CDC_RESOLVED_TS_ADVANCE_METHOD.get(), 0);
    }
    suite
        .cluster
        .pd_client()
        .feature_gate()
        .set_version("5.0.0")
        .unwrap();

    loop {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            assert!(resolved_ts.regions == vec![region.id]);
            if CDC_RESOLVED_TS_ADVANCE_METHOD.get() == 1 {
                break;
            }
        }
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
// I cannot fix this case yet...
#[cfg(NO_NEXT_GEN_COMPATIBLE)]
fn test_resolved_ts_with_learners() {
    let mut cluster = new_server_cluster(0, 2);
    cluster.pd_client.disable_default_operator();
    let r = cluster.run_conf_change();
    cluster.pd_client.must_add_peer(r, new_learner_peer(2, 2));
    let mut suite = build_testsuite_from_cluster(cluster);

    let region = get_ready_region(&mut suite, &[]);
    let rid = region.id;
    let req = suite.new_changedata_request(rid);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(rid));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_resolved_ts_with_learners initial subscribe",
    );

    for _ in 0..10 {
        let event = receive_event(true);
        if event.has_resolved_ts() {
            assert!(event.get_resolved_ts().regions == vec![rid]);
            drop(receive_event);
            suite.stop();
            return;
        }
    }
    panic!("resolved timestamp should be advanced correctly");
}

#[test]
fn test_prewrite_without_value() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(2)
        .after_cluster_bootstrapped(|c| c.cluster.get_pd_client().disable_default_operator())
        .build();
    let region = get_ready_region(&mut suite, b"xkey");
    let rid = region.id;
    let ctx = suite.get_context(rid);
    let client = suite.get_tikv_client(rid).clone();
    let large_value = vec![b'x'; 2 * txn_types::SHORT_VALUE_MAX_LEN];

    // Perform a pessimistic prewrite with a large value.
    let mut muts = vec![Mutation::default()];
    muts[0].set_op(Op::Put);
    muts[0].key = b"xkeykey".to_vec();
    muts[0].value = large_value.clone();
    let w = try_kv_prewrite_pessimistic(&client, ctx.clone(), muts, b"xkeykey".to_vec(), 10);
    assert!(w.get_errors().is_empty(), "{:?}", w);

    let req = suite.new_changedata_request(rid);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(rid));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_prewrite_without_value initial subscribe",
    );

    // check_txn_status will put the lock again, but without value.
    must_check_txn_status(&client, ctx.clone(), b"xkeykey", 10, 12, 12);
    must_kv_commit(&client, ctx, vec![b"xkeykey".to_vec()], 10, 14, 14);
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        // The lock without value shouldn't be retrieved.
        // Cloud Engine: value is always inlined in `Lock`.
        let event = receive_event(false);
        assert!(
            !event.get_events()[0].get_entries().entries[0]
                .get_value()
                .is_empty(),
            "{:?}",
            event
        );
    }
    let event = receive_event(false);
    assert_eq!(event.get_events()[0].get_entries().entries[0].commit_ts, 14);
}

#[test]
#[ignore = "txn_source isn't supported by kvengine yet; maybe add them to `user_meta`"]
fn test_filter_loop() {
    test_kv_format_impl!(test_filter_loop_impl<ApiV2>);
}

fn test_filter_loop_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let region = get_ready_region(&mut suite, b"xk");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    req.set_filter_loop(true);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_filter_loop initial subscribe",
    );

    // Insert value, simulate INSERT INTO.
    let mut m1 = Mutation::default();
    let k1 = b"xk1".to_vec();
    m1.set_op(Op::Insert);
    m1.key = k1.clone();
    m1.value = b"v1".to_vec();
    suite.must_kv_prewrite_with_source(region_id, vec![m1], k1.clone(), 10.into(), 1);
    let mut m2 = Mutation::default();
    let k2 = b"xk2".to_vec();
    m2.set_op(Op::Insert);
    m2.key = k2.clone();
    m2.value = b"v2".to_vec();
    suite.must_kv_prewrite_with_source(region_id, vec![m2], k2.clone(), 12.into(), 0);
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let events = es.take_entries().to_vec();
            assert_eq!(events.len(), 1);
            let row = &events[0];
            assert_eq!(row.get_value(), b"v2");
            assert_eq!(row.get_old_value(), b"");
            assert_eq!(row.get_type(), EventLogType::Prewrite);
            assert_eq!(row.get_start_ts(), 12);
        }
        other => panic!("unknown event {:?}", other),
    }
    suite.must_kv_commit_with_source(region_id, vec![k1], 10.into(), 15.into(), 1);
    suite.must_kv_commit_with_source(region_id, vec![k2], 12.into(), 17.into(), 0);
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let events = es.take_entries().to_vec();
            assert_eq!(events.len(), 1);
            let row = &events[0];
            assert_eq!(row.get_type(), EventLogType::Commit);
            assert_eq!(row.get_commit_ts(), 17);
        }
        other => panic!("unknown event {:?}", other),
    }

    // Rollback
    let mut m3 = Mutation::default();
    let k3 = b"xk3".to_vec();
    m3.set_op(Op::Put);
    m3.key = k3.clone();
    m3.value = b"v3".to_vec();
    suite.must_kv_prewrite_with_source(region_id, vec![m3], k3.clone(), 30.into(), 1);
    suite.must_kv_rollback(region_id, vec![k3], 30.into());
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let events = es.take_entries().to_vec();
            assert_eq!(events.len(), 1);
            let row = &events[0];
            assert_eq!(row.get_type(), EventLogType::Rollback);
            assert_eq!(row.get_commit_ts(), 0);
        }
        other => panic!("unknown event {:?}", other),
    }

    // Update value
    let k1 = b"xk1".to_vec();
    let mut m4 = Mutation::default();
    m4.set_op(Op::Put);
    m4.key = k1.clone();
    m4.value = vec![b'3'; 5120];
    suite.must_kv_prewrite_with_source(region_id, vec![m4], k1.clone(), 40.into(), 1);
    suite.must_kv_commit_with_source(region_id, vec![k1], 40.into(), 42.into(), 1);
    let k2 = b"xk2".to_vec();
    let mut m5 = Mutation::default();
    m5.set_op(Op::Put);
    m5.key = k2.clone();
    m5.value = vec![b'4'; 5121];
    suite.must_kv_prewrite(region_id, vec![m5], k2.clone(), 44.into());
    suite.must_kv_commit(region_id, vec![k2.clone()], 44.into(), 46.into());
    let mut events = receive_event(false).events.to_vec();
    if events.len() == 1 {
        events.extend(receive_event(false).events.into_iter());
    }
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let events = es.take_entries().to_vec();
            assert_eq!(events.len(), 1);
            assert_eq!(events[0].get_type(), EventLogType::Prewrite);
            assert_eq!(events[0].get_start_ts(), 44);
            assert_eq!(events[0].get_key(), k2.as_slice());
        }
        other => panic!("unknown event {:?}", other),
    }
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let events = es.take_entries().to_vec();
            assert_eq!(events.len(), 1);
            assert_eq!(events[0].get_type(), EventLogType::Commit);
            assert_eq!(events[0].get_commit_ts(), 46);
            assert_eq!(events[0].get_key(), k2.as_slice());
        }
        other => panic!("unknown event {:?}", other),
    }

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
#[ignore = "This case was written for APIv1."]
fn test_flashback() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(1)
        .cfg_fun(|_id, cfg| cfg.resolved_ts.advance_ts_interval = ReadableDuration::millis(50))
        .build();

    let key = Key::from_raw(b"xkeya");
    let region = get_ready_region(&mut suite, key.as_encoded());
    let region_id = region.get_id();
    let req = suite.new_changedata_request(region_id);
    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_flashback initial subscribe",
    );
    // Sleep a while to make sure the stream is registered.
    sleep_ms(1000);
    let start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    for i in 0..2 {
        let (k, v) = (
            format!("xkeykey{}", i).as_bytes().to_vec(),
            format!("value{}", i).as_bytes().to_vec(),
        );
        // Prewrite
        let start_ts1 = block_on(suite.cluster.pd_client().get_tso()).unwrap();
        let mut mutation = Mutation::default();
        mutation.set_op(Op::Put);
        mutation.key = k.clone();
        mutation.value = v;
        suite.must_kv_prewrite(region_id, vec![mutation], k.clone(), start_ts1);
        // Commit
        let commit_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
        suite.must_kv_commit(region_id, vec![k.clone()], start_ts1, commit_ts);
    }
    let (start_key, end_key) = (b"xkeykey0".to_vec(), b"xkeykey2".to_vec());
    // Prepare flashback.
    let flashback_start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_prepare_flashback(region_id, &start_key, &end_key, flashback_start_ts);
    // resolved ts should not be advanced anymore.
    let mut counter = 0;
    let mut last_resolved_ts = 0;
    loop {
        let event = receive_event(true);
        if let Some(resolved_ts) = event.resolved_ts.as_ref() {
            if resolved_ts.ts == last_resolved_ts {
                counter += 1;
            }
            last_resolved_ts = resolved_ts.ts;
        }
        if counter > 20 {
            break;
        }
        sleep_ms(50);
    }
    // Flashback.
    let flashback_commit_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    suite.must_kv_flashback(
        region_id,
        &start_key,
        &end_key,
        flashback_start_ts,
        flashback_commit_ts,
        start_ts,
    );
    // Check the flashback event.
    let mut resolved_ts = 0;
    let mut event_counter = 0;
    loop {
        let mut cde = receive_event(true);
        if cde.get_resolved_ts().get_ts() > resolved_ts {
            resolved_ts = cde.get_resolved_ts().get_ts();
        }
        let events = cde.mut_events();
        if !events.is_empty() {
            assert_eq!(events.len(), 1);
            match events.pop().unwrap().event.unwrap() {
                Event_oneof_event::Entries(entries) => {
                    assert_eq!(entries.entries.len(), 1);
                    event_counter += 1;
                    let e = &entries.entries[0];
                    assert!(e.commit_ts > resolved_ts);
                    assert_eq!(e.get_op_type(), EventRowOpType::Delete);
                    match e.get_type() {
                        EventLogType::Committed => {
                            // First entry should be a 1PC flashback.
                            assert_eq!(e.get_key(), b"xkeykey1");
                            assert_eq!(event_counter, 1);
                        }
                        EventLogType::Commit => {
                            // Second entry should be a 2PC commit.
                            assert_eq!(e.get_key(), b"xkeykey0");
                            assert_eq!(event_counter, 2);
                            break;
                        }
                        _ => panic!("unknown event type {:?}", e.get_type()),
                    }
                }
                other => panic!("unknown event {:?}", other),
            }
        }
    }
}

#[test]
fn test_cdc_filter_key_range() {
    let mut suite = TestSuite::new(1, ApiVersion::V2);
    let region = get_ready_region(&mut suite, b"xkey0");
    let region_id = region.id;

    let req = suite.new_changedata_request(region_id);

    // Observe range [key1, key3).
    let mut req_1_3 = req.clone();
    req_1_3.request_id = 13;
    req_1_3.start_key = Key::from_raw(b"xkeykey1").into_encoded();
    req_1_3.end_key = Key::from_raw(b"xkeykey3").into_encoded();
    let (mut req_tx13, _event_feed_wrap13, receive_event13) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx13,
        receive_event13.as_ref(),
        &req_1_3,
        "test_cdc_filter_key_range [key1,key3) subscribe",
    );

    let (mut req_tx24, _event_feed_wrap24, receive_event24) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    let mut req_2_4 = req;
    req_2_4.request_id = 24;
    req_2_4.start_key = Key::from_raw(b"xkeykey2").into_encoded();
    req_2_4.end_key = Key::from_raw(b"xkeykey4").into_encoded();
    subscribe_until_initialized(
        &mut req_tx24,
        receive_event24.as_ref(),
        &req_2_4,
        "test_cdc_filter_key_range [key2,key4) subscribe",
    );

    // Sleep a while to make sure the stream is registered.
    sleep_ms(1000);

    let receive_and_check_events = |is13: bool, is24: bool| -> Vec<Event> {
        if is13 && is24 {
            let mut events = receive_event13(false).events.to_vec();
            let mut events24 = receive_event24(false).events.to_vec();
            events.append(&mut events24);
            events
        } else if is13 {
            let events = receive_event13(false).events.to_vec();
            let event = receive_event24(true);
            assert!(event.resolved_ts.is_some(), "{:?}", event);
            events
        } else if is24 {
            let events = receive_event24(false).events.to_vec();
            let event = receive_event13(true);
            assert!(event.resolved_ts.is_some(), "{:?}", event);
            events
        } else {
            let event = receive_event13(true);
            assert!(event.resolved_ts.is_some(), "{:?}", event);
            let event = receive_event24(true);
            assert!(event.resolved_ts.is_some(), "{:?}", event);
            vec![]
        }
    };
    for case in &[
        ("xkeykey1", true, false, true /* commit */),
        ("xkeykey1", true, false, false /* rollback */),
        ("xkeykey2", true, true, true),
        ("xkeykey3", false, true, true),
        ("xkeykey4", false, false, true),
    ] {
        let (k, v) = (case.0.to_owned(), "value".to_owned());
        // Prewrite
        let start_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
        let mut mutation = Mutation::default();
        mutation.set_op(Op::Put);
        mutation.key = k.clone().into_bytes();
        mutation.value = v.into_bytes();
        suite.must_kv_prewrite(region_id, vec![mutation], k.clone().into_bytes(), start_ts);

        if case.3 {
            // Commit
            let commit_ts = block_on(suite.cluster.pd_client().get_tso()).unwrap();
            suite.must_kv_commit(region_id, vec![k.into_bytes()], start_ts, commit_ts);
            let mut events = receive_and_check_events(case.1, case.2);
            while let Some(event) = events.pop() {
                match event.event.unwrap() {
                    Event_oneof_event::Entries(entries) => {
                        assert_eq!(entries.entries.len(), 1);
                        assert_eq!(entries.entries[0].get_type(), EventLogType::Committed);
                    }
                    other => panic!("unknown event {:?}", other),
                }
            }
        } else {
            // Rollback
            suite.must_kv_rollback(region_id, vec![k.into_bytes()], start_ts);
            let mut events = receive_and_check_events(case.1, case.2);
            while let Some(event) = events.pop() {
                match event.event.unwrap() {
                    Event_oneof_event::Entries(entries) => {
                        assert_eq!(entries.entries.len(), 1);
                        assert_eq!(entries.entries[0].get_type(), EventLogType::Rollback);
                    }
                    other => panic!("unknown event {:?}", other),
                }
            }
        }
    }

    suite.stop();
}

#[test]
fn test_cdc_partial_subscription() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(1)
        .cfg_fun(|_id, cfg| {
            configure_for_lease_read_cfg(cfg, Some(100), Some(10));
        })
        .after_cluster_bootstrapped(|cluster| {
            cluster.get_pd_client().disable_default_operator();
        })
        .build();
    let region = get_ready_region(&mut suite, b"xkey");
    let rid = region.id;

    let prewrite_tso = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let (k, v) = (b"xkey0".to_vec(), vec![b'x'; 16]);
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = k.clone();
    mutation.value = v;
    suite.must_kv_prewrite(rid, vec![mutation], k.clone(), prewrite_tso);

    let cf_tso = block_on(suite.cluster.pd_client().get_tso()).unwrap();
    let (mut req_tx, _, receive_event) = new_event_feed_v2(suite.get_region_cdc_client(rid));
    let mut req = suite.new_changedata_request(rid);
    req.request_id = 1;
    req.checkpoint_ts = cf_tso.into_inner();
    req.set_start_key(Key::from_raw(b"y").into_encoded());
    req.set_end_key(Key::from_raw(b"z").into_encoded());
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_partial_subscription initial subscribe",
    );

    // Resolved ts may only advance one logical tick per min-ts interval, so
    // partial subscriptions need a wider observation window than the original
    // 1s loop on this branch.
    for _ in 0..300 {
        let cdc_event = receive_event(true);
        if cdc_event.has_resolved_ts() {
            let resolved_ts = cdc_event.get_resolved_ts();
            if resolved_ts.ts > prewrite_tso.into_inner() {
                return;
            }
        }
    }
    panic!("resolved_ts should exceed prewrite_tso");
}

#[test]
fn test_cdc_rollback_prewrites_with_txn_source() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(1)
        .cfg_fun(|_id, cfg| {
            configure_for_lease_read_cfg(cfg, Some(100), Some(10));
        })
        .presplit_magic_keyspace()
        .after_cluster_bootstrapped(|cluster| {
            cluster.get_pd_client().disable_default_operator();
        })
        .build();
    let region = get_ready_region(&mut suite, b"xkey0");
    let rid = region.id;
    let cf_tso = block_on(suite.cluster.pd_client().get_tso()).unwrap();

    let (mut req_tx, _, receive_event) = new_event_feed_v2(suite.get_region_cdc_client(rid));
    let mut req = suite.new_changedata_request(rid);
    req.request_id = 1;
    req.checkpoint_ts = cf_tso.into_inner();
    req.filter_loop = true;
    req.set_start_key(Key::from_raw(b"xkeya").into_encoded());
    req.set_end_key(Key::from_raw(b"xkeyz").into_encoded());
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_rollback_prewrites_with_txn_source initial subscribe",
    );

    let start_tso = cf_tso.next();
    let k = b"xkeykey".to_vec();
    let v = vec![b'x'; 16 * 1024];
    let mut mutation = Mutation::default();
    mutation.set_op(Op::Put);
    mutation.key = k.clone();
    mutation.value = v;
    suite.must_kv_prewrite_with_source(rid, vec![mutation], k.clone(), start_tso, 1);

    loop {
        let cdc_event = receive_event(true);
        if cdc_event.has_resolved_ts() {
            let resolved_ts = cdc_event.get_resolved_ts().get_ts();
            assert!(
                resolved_ts <= start_tso.into_inner(),
                "resolved ts advanced past hidden prewrite: {resolved_ts} > {}",
                start_tso.into_inner()
            );
            break;
        }
    }

    suite.must_kv_rollback(rid, vec![k.clone()], start_tso);

    // We can't receive the prewrite because it's with a txn_source,
    // but we can receive the rollback.
    let mut rollbacked = false;
    let timer = Instant::now();
    loop {
        let cdc_event = receive_event(true);
        for event in cdc_event.get_events() {
            for entry in event.get_entries().get_entries() {
                match entry.get_type() {
                    EventLogType::Rollback => rollbacked = true,
                    _ => unreachable!(),
                }
            }
        }
        if rollbacked && cdc_event.get_resolved_ts().get_ts() > start_tso.into_inner() {
            return;
        }
        assert!(
            timer.elapsed() <= Duration::from_secs(15),
            "resolved ts did not advance after rollback"
        );
    }
}

#[test]
fn test_cdc_pessimistic_lock_unlock() {
    let mut suite = CloudTestSuiteBuilder::new()
        .num_nodes(1)
        .cfg_fun(|_id, cfg| {
            configure_for_lease_read_cfg(cfg, Some(100), Some(10));
        })
        .after_cluster_bootstrapped(|c| c.cluster.get_pd_client().disable_default_operator())
        .build();
    let region = get_ready_region(&mut suite, &[]);
    let rid = region.id;
    let cf_tso = block_on(suite.cluster.pd_client().get_tso()).unwrap();

    let (mut req_tx, _, receive_event) = new_event_feed_v2(suite.get_region_cdc_client(rid));
    let mut req = suite.new_changedata_request(rid);
    req.request_id = 1;
    req.checkpoint_ts = cf_tso.into_inner();
    req.filter_loop = true;
    req.set_start_key(Key::from_raw(b"xkeya").into_encoded());
    req.set_end_key(Key::from_raw(b"xkeyz").into_encoded());
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_cdc_pessimistic_lock_unlock initial subscribe",
    );

    let start_tso = cf_tso.next();
    let k = b"xkeykey".to_vec();
    let v = vec![b'x'; 16 * 1024];
    let mut mutation = Mutation::default();
    mutation.set_op(Op::PessimisticLock);
    mutation.key = k.clone();
    mutation.value = v;
    let for_update_tso = TimeStamp::from(start_tso.into_inner() + 10);
    suite.must_acquire_pessimistic_lock(rid, vec![mutation], k.clone(), start_tso, for_update_tso);
    std::thread::sleep(Duration::from_millis(500));

    suite.must_release_pessimistic_lock(rid, k.clone(), start_tso, for_update_tso);
    std::thread::sleep(Duration::from_millis(500));
}