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

use std::{
    collections::btree_map::{BTreeMap, Entry as BTreeMapEntry},
    fmt,
    ops::Bound,
    result::Result as StdResult,
    string::String,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    },
    time::Duration,
};

use api_version::ApiV2;
use collections::HashMap;
use crossbeam::atomic::AtomicCell;
use kvengine::{
    IoType, Iterator as KvIterator, ToEncryptionState, UserMeta,
    table::{
        InnerKey,
        txn_file::{SkipOpTxnFileIterator, TxnFileIterator},
    },
    util::TxnFileRefExt,
};
use kvproto::{
    cdcpb::{
        Error as EventError, Event, Event_oneof_event, EventEntries, EventLogType, EventRow,
        EventRowOpType,
    },
    kvrpcpb::ExtraOp as TxnExtraOp,
    metapb::{Region, RegionEpoch},
    raft_cmdpb::{AdminCmdType, AdminRequest, AdminResponse, CmdType, PutRequest, Request},
};
use raftstore::{
    Error as RaftStoreError,
    coprocessor::{CdcExtra, Cmd, CmdBatch, ObserveHandle, TxnFileCdcRecord},
    store::util::compare_region_epoch,
};
use tikv::storage::{Statistics, txn::TxnEntry};
use tikv_util::{
    box_err, debug, info,
    memory::{HeapSize, MemoryQuota, OwnedAllocated},
    time::Instant,
    warn,
};
use txn_types::{Key, Lock, LockType, TimeStamp, WriteBatchFlags, WriteRef, WriteType};

use crate::{
    Error, Result,
    channel::{CDC_EVENT_MAX_BYTES, CdcEvent, SendError, Sink},
    endpoint::{Advance, TxnFileLoaderContext},
    metrics::*,
    old_value::{OldValueCache, OldValueCallback},
    service::{ConnId, RequestId},
    txn_source::TxnSource,
};

static DOWNSTREAM_ID_ALLOC: AtomicUsize = AtomicUsize::new(0);

/// A unique identifier of a Downstream.
#[derive(Clone, Copy, Debug, PartialEq, Hash)]
pub struct DownstreamId(usize);

impl DownstreamId {
    pub fn new() -> DownstreamId {
        DownstreamId(DOWNSTREAM_ID_ALLOC.fetch_add(1, Ordering::SeqCst))
    }
}

impl Default for DownstreamId {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub enum DownstreamState {
    /// It's just created and rejects change events and resolved timestamps.
    #[default]
    Uninitialized,
    /// It has got a snapshot for incremental scan, and change events will be
    /// accepted. However, it still rejects resolved timestamps.
    Initializing,
    /// Incremental scan is finished so that resolved timestamps are acceptable
    /// now.
    Normal,
    Stopped,
}

/// Should only be called when it's uninitialized or stopped. Return false if
/// it's stopped.
pub(crate) fn on_init_downstream(s: &AtomicCell<DownstreamState>) -> bool {
    s.compare_exchange(
        DownstreamState::Uninitialized,
        DownstreamState::Initializing,
    )
    .is_ok()
}

/// Should only be called when it's initializing or stopped. Return false if
/// it's stopped.
pub(crate) fn post_init_downstream(s: &AtomicCell<DownstreamState>) -> bool {
    s.compare_exchange(DownstreamState::Initializing, DownstreamState::Normal)
        .is_ok()
}

impl DownstreamState {
    /// Returns whether a removed downstream may still own a queued scan.
    ///
    /// A queued scan is uninitialized, while lock-scan validation can mark it
    /// stopped before deregistration removes the downstream.
    fn may_have_queued_scan(self) -> bool {
        match self {
            DownstreamState::Uninitialized | DownstreamState::Stopped => true,
            DownstreamState::Initializing | DownstreamState::Normal => false,
        }
    }

    pub fn ready_for_change_events(&self) -> bool {
        match *self {
            DownstreamState::Uninitialized | DownstreamState::Stopped => false,
            DownstreamState::Initializing | DownstreamState::Normal => true,
        }
    }

    pub fn ready_for_advancing_ts(&self) -> bool {
        match *self {
            DownstreamState::Normal => true,

            DownstreamState::Uninitialized
            | DownstreamState::Stopped
            | DownstreamState::Initializing => false,
        }
    }
}

/// State needed by the endpoint after removing one downstream.
pub(crate) struct UnsubscribeResult {
    pub(crate) delegate_empty: bool,
    pub(crate) may_have_queued_scan: bool,
}

pub struct Downstream {
    /// A unique identifier of the Downstream.
    pub id: DownstreamId,
    /// The IP address of downstream.
    pub peer: String,
    pub region_epoch: RegionEpoch,
    /// The request ID set by CDC to identify events corresponding different
    /// requests.
    pub req_id: RequestId,
    pub conn_id: ConnId,

    pub filter_loop: bool,
    pub observed_range: ObservedRange,

    sink: Option<Sink>,
    state: Arc<AtomicCell<DownstreamState>>,
    pub(crate) scan_truncated: Arc<AtomicBool>,

    // Fields to handle ResolvedTs advancing. If `lock_heap` is none it means
    // the downstream hasn't finished the incremental scanning.
    lock_heap: Option<BTreeMap<TimeStamp, isize>>,
    advanced_to: TimeStamp,
}

impl fmt::Debug for Downstream {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Downstream")
            .field("id", &self.id)
            .field("request_id", &self.req_id)
            .field("conn_id", &self.conn_id)
            .finish()
    }
}

impl Downstream {
    /// Create a Downstream.
    ///
    /// peer is the address of the downstream.
    /// sink sends data to the downstream.
    pub fn new(
        peer: String,
        region_epoch: RegionEpoch,
        req_id: RequestId,
        conn_id: ConnId,
        filter_loop: bool,
        observed_range: ObservedRange,
    ) -> Downstream {
        Downstream {
            id: DownstreamId::new(),
            peer,
            region_epoch,
            req_id,
            conn_id,
            filter_loop,

            observed_range,

            sink: None,
            state: Arc::new(AtomicCell::new(DownstreamState::default())),
            scan_truncated: Arc::new(AtomicBool::new(false)),

            lock_heap: None,
            advanced_to: TimeStamp::zero(),
        }
    }

    // NOTE: it's not allowed to sink `EventError` directly by this function,
    // because the sink can be also used by an incremental scan. We must ensure
    // no more events can be pushed to the sink after an `EventError` is sent.
    pub fn sink_event(&self, mut event: Event, force: bool) -> Result<()> {
        event.set_request_id(self.req_id.0);
        if self.sink.is_none() {
            info!("cdc drop event, no sink"; "downstream_id" => ?self.id,
                "request_id" => ?self.req_id, "conn_id" => ?self.conn_id);
            return Err(Error::Sink(SendError::Disconnected));
        }
        let sink = self.sink.as_ref().unwrap();
        match sink.unbounded_send(CdcEvent::Event(event), force) {
            Ok(_) => Ok(()),
            Err(SendError::Disconnected) => {
                debug!("cdc send event failed, disconnected"; "downstream_id" => ?self.id,
                    "request_id" => ?self.req_id, "conn_id" => ?self.conn_id);
                Err(Error::Sink(SendError::Disconnected))
            }
            // TODO handle errors.
            Err(e @ SendError::Full) | Err(e @ SendError::Congested) => {
                info!("cdc send event failed, full"; "downstream_id" => ?self.id,
                    "request_id" => ?self.req_id, "conn_id" => ?self.conn_id);
                Err(Error::Sink(e))
            }
        }
    }

    /// EventErrors must be sent by this function. And we must ensure no more
    /// events or ResolvedTs will be sent to the downstream after
    /// `sink_error_event` is called.
    pub fn sink_error_event(&self, region_id: u64, err_event: EventError) -> Result<()> {
        self.scan_truncated.store(true, Ordering::Release);
        let mut change_data_event = Event::default();
        change_data_event.event = Some(Event_oneof_event::Error(err_event));
        change_data_event.region_id = region_id;
        // Try it's best to send error events.
        let force_send = true;
        self.sink_event(change_data_event, force_send)
    }

    pub fn set_sink(&mut self, sink: Sink) {
        self.sink = Some(sink);
    }

    pub fn get_state(&self) -> Arc<AtomicCell<DownstreamState>> {
        self.state.clone()
    }
}

// In `PendingLock`,  `key` is encoded.
pub enum PendingLock {
    Track { key: Key, start_ts: MiniLock },
    Untrack { key: Key, start_ts: TimeStamp },
}

impl HeapSize for PendingLock {
    fn approximate_heap_size(&self) -> usize {
        match self {
            PendingLock::Track { key, .. } | PendingLock::Untrack { key, .. } => {
                key.approximate_heap_size()
            }
        }
    }
}

pub enum LockTracker {
    Pending,
    Preparing(Vec<PendingLock>),
    Prepared {
        region: Region,
        locks: BTreeMap<Key, MiniLock>,
    },
}

impl fmt::Debug for LockTracker {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            LockTracker::Pending => write!(f, "LockTracker::Pending"),
            LockTracker::Preparing(ref locks) => {
                write!(f, "LockTracker::Preparing({})", locks.len())
            }
            LockTracker::Prepared { locks, .. } => {
                write!(f, "LockTracker::Prepared({})", locks.len())
            }
        }
    }
}

/// `MiniLock` is like `Lock`, but only contains fields that CDC cares about.
#[derive(Eq, PartialEq, Debug)]
pub struct MiniLock {
    pub ts: TimeStamp,
    pub txn_source: u64,
}

impl MiniLock {
    pub fn new<T>(ts: T, txn_source: u64) -> Self
    where
        TimeStamp: From<T>,
    {
        MiniLock {
            ts: TimeStamp::from(ts),
            txn_source,
        }
    }

    #[cfg(test)]
    pub fn from_ts<T>(ts: T) -> Self
    where
        TimeStamp: From<T>,
    {
        MiniLock {
            ts: TimeStamp::from(ts),
            txn_source: 0,
        }
    }
}

/// A CDC delegate of a raftstore region peer.
///
/// It converts raft commands into CDC events and broadcast to downstreams.
/// It also tracks transactions on the fly in order to compute resolved ts.
pub struct Delegate {
    pub region_id: u64,
    pub handle: ObserveHandle,
    memory_quota: Arc<MemoryQuota>,

    lock_tracker: LockTracker,
    downstreams: Vec<Downstream>,
    txn_extra_op: Arc<AtomicCell<TxnExtraOp>>,
    failed: bool,

    created: Instant,
    last_lag_warn: Instant,
}

impl Drop for Delegate {
    fn drop(&mut self) {
        match &self.lock_tracker {
            LockTracker::Pending => {}
            LockTracker::Preparing(locks) => {
                let mut free_bytes = 0;
                for lock in locks {
                    free_bytes += lock.approximate_heap_size();
                }
                self.memory_quota.free(free_bytes);
                CDC_PENDING_BYTES_GAUGE.sub(free_bytes as _);
            }
            LockTracker::Prepared { locks, .. } => {
                let mut free_bytes = 0;
                for lock in locks.keys() {
                    free_bytes += lock.approximate_heap_size();
                }
                self.memory_quota.free(free_bytes);
                CDC_PENDING_BYTES_GAUGE.sub(free_bytes as _);
            }
        }
    }
}

impl Delegate {
    fn push_lock(&mut self, key: Key, start_ts: MiniLock) -> Result<isize> {
        let bytes = key.approximate_heap_size();
        let mut lock_count_modify = 0;
        match &mut self.lock_tracker {
            LockTracker::Pending => unreachable!(),
            LockTracker::Preparing(locks) => {
                self.memory_quota.alloc(bytes)?;
                CDC_PENDING_BYTES_GAUGE.add(bytes as _);
                locks.push(PendingLock::Track { key, start_ts });
            }
            LockTracker::Prepared { locks, .. } => {
                if locks.insert(key, start_ts).is_none() {
                    self.memory_quota.alloc(bytes)?;
                    CDC_PENDING_BYTES_GAUGE.add(bytes as _);
                    lock_count_modify = 1;
                }
            }
        }
        Ok(lock_count_modify)
    }

    fn pop_lock(&mut self, key: Key, start_ts: TimeStamp) -> Result<isize> {
        let mut lock_count_modify = 0;
        match &mut self.lock_tracker {
            LockTracker::Pending => unreachable!(),
            LockTracker::Preparing(locks) => {
                let bytes = key.approximate_heap_size();
                self.memory_quota.alloc(bytes)?;
                CDC_PENDING_BYTES_GAUGE.add(bytes as _);
                locks.push(PendingLock::Untrack { key, start_ts });
            }
            LockTracker::Prepared { locks, .. } => {
                if let BTreeMapEntry::Occupied(x) = locks.entry(key) {
                    if x.get().ts == start_ts {
                        let (key, _) = x.remove_entry();
                        let bytes = key.approximate_heap_size();
                        self.memory_quota.free(bytes);
                        CDC_PENDING_BYTES_GAUGE.sub(bytes as _);
                        lock_count_modify = -1;
                    }
                }
            }
        }
        Ok(lock_count_modify)
    }

    pub(crate) fn init_lock_tracker(&mut self) -> bool {
        if matches!(self.lock_tracker, LockTracker::Pending) {
            self.lock_tracker = LockTracker::Preparing(vec![]);
            return true;
        }
        false
    }

    fn finish_prepare_lock_tracker(
        &mut self,
        region: Region,
        mut locks: BTreeMap<Key, MiniLock>,
    ) -> Result<()> {
        let delta_locks = match std::mem::replace(&mut self.lock_tracker, LockTracker::Pending) {
            LockTracker::Preparing(locks) => locks,
            _ => unreachable!(),
        };

        let mut free_bytes = 0usize;
        for delta_lock in delta_locks {
            free_bytes += delta_lock.approximate_heap_size();
            match delta_lock {
                PendingLock::Track { key, start_ts } => match locks.entry(key) {
                    BTreeMapEntry::Vacant(x) => {
                        x.insert(start_ts);
                    }
                    BTreeMapEntry::Occupied(x) => {
                        assert_eq!(x.get().ts, start_ts.ts);
                    }
                },
                PendingLock::Untrack { key, start_ts } => {
                    if let BTreeMapEntry::Occupied(x) = locks.entry(key) {
                        if x.get().ts == start_ts {
                            x.remove();
                        }
                    }
                }
            }
        }
        self.memory_quota.free(free_bytes);
        CDC_PENDING_BYTES_GAUGE.sub(free_bytes as _);

        let mut alloc_bytes = 0usize;
        for key in locks.keys() {
            alloc_bytes += key.approximate_heap_size();
        }
        self.memory_quota.alloc(alloc_bytes)?;
        CDC_PENDING_BYTES_GAUGE.add(alloc_bytes as _);
        self.lock_tracker = LockTracker::Prepared { region, locks };
        Ok(())
    }

    pub(crate) fn finish_scan_locks(
        &mut self,
        region: Region,
        locks: BTreeMap<Key, MiniLock>,
    ) -> Result<Vec<(&Downstream, Error)>> {
        fail::fail_point!("cdc_finish_scan_locks_memory_quota_exceed", |_| Err(
            Error::MemoryQuotaExceeded(tikv_util::memory::MemoryQuotaExceeded)
        ));

        self.finish_prepare_lock_tracker(region, locks)?;
        info!("cdc region is ready"; "region_id" => self.region_id);

        let region = match &self.lock_tracker {
            LockTracker::Prepared { region, .. } => region,
            _ => unreachable!(),
        };

        // Check observed key range in region.
        let mut failed_downstreams = Vec::new();
        for downstream in &mut self.downstreams {
            downstream.observed_range.update_region_key_range(region);
            if let Err(e) = Self::check_epoch_on_ready(downstream, region) {
                failed_downstreams.push((&*downstream, e));
            }
        }

        Ok(failed_downstreams)
    }

    /// Create a Delegate the given region.
    pub fn new(
        region_id: u64,
        memory_quota: Arc<MemoryQuota>,
        txn_extra_op: Arc<AtomicCell<TxnExtraOp>>,
    ) -> Delegate {
        Delegate {
            region_id,
            handle: ObserveHandle::new(),
            memory_quota,

            lock_tracker: LockTracker::Pending,
            downstreams: Vec::new(),
            txn_extra_op,
            failed: false,

            created: Instant::now_coarse(),
            last_lag_warn: Instant::now_coarse(),
        }
    }

    /// Let downstream subscribe the delegate.
    /// Return error if subscribe fails and the `Delegate` won't be changed.
    pub fn subscribe(&mut self, downstream: Downstream) -> StdResult<(), (Error, Downstream)> {
        if let LockTracker::Prepared { ref region, .. } = &self.lock_tracker {
            // Check if the downstream is outdated.
            if let Err(e) = Self::check_epoch_on_ready(&downstream, region) {
                return Err((e, downstream));
            }
        }
        self.add_downstream(downstream);
        Ok(())
    }

    pub fn downstream(&self, downstream_id: DownstreamId) -> Option<&Downstream> {
        self.downstreams().iter().find(|d| d.id == downstream_id)
    }

    pub fn downstreams(&self) -> &Vec<Downstream> {
        &self.downstreams
    }

    pub fn downstreams_mut(&mut self) -> &mut Vec<Downstream> {
        &mut self.downstreams
    }

    /// Let downstream unsubscribe the delegate.
    /// Return whether the delegate is empty or not.
    pub fn unsubscribe(&mut self, id: DownstreamId, err: Option<Error>) -> bool {
        self.unsubscribe_with_scan_cleanup(id, err).delegate_empty
    }

    /// Unsubscribes one downstream and reports whether scan queue cleanup may
    /// be needed.
    pub(crate) fn unsubscribe_with_scan_cleanup(
        &mut self,
        id: DownstreamId,
        err: Option<Error>,
    ) -> UnsubscribeResult {
        let error_event = err.map(|err| err.into_error_event(self.region_id));
        let region_id = self.region_id;
        let mut may_have_queued_scan = false;
        if let Some(d) = self.remove_downstream(id) {
            if let Some(error_event) = error_event {
                info!("cdc unsubscribes downstream due to error";
                    "error" => ?error_event, "downstream" => ?d.peer,
                    "downstream_id" => ?d.id, "request_id" => ?d.req_id,
                    "region_id" => region_id, "conn_id" => ?d.conn_id);
                if let Err(err) = d.sink_error_event(region_id, error_event.clone()) {
                    warn!("cdc send unsubscribe failed";
                        "error" => ?err, "origin_error" => ?error_event,
                        "downstream_id" => ?d.id, "downstream" => ?d.peer,
                        "request_id" => ?d.req_id, "region_id" => region_id, "conn_id" => ?d.conn_id);
                }
            }
            may_have_queued_scan = d
                .state
                .swap(DownstreamState::Stopped)
                .may_have_queued_scan();
        }
        UnsubscribeResult {
            delegate_empty: self.downstreams().is_empty(),
            may_have_queued_scan,
        }
    }

    pub fn mark_failed(&mut self) {
        self.failed = true;
    }

    pub fn has_failed(&self) -> bool {
        self.failed
    }

    /// Stop the delegate
    ///
    /// This means the region has met an unrecoverable error for CDC.
    /// It broadcasts errors to all downstream and stops.
    pub fn stop(&mut self, err: Error) {
        self.stop_with_scan_cleanup(err);
    }

    /// Stops all downstreams and reports whether scan queue cleanup may be
    /// needed.
    pub(crate) fn stop_with_scan_cleanup(&mut self, err: Error) -> bool {
        self.mark_failed();
        self.stop_observing();
        let region_id = self.region_id;
        let error = err.into_error_event(self.region_id);
        let mut may_have_queued_scan = false;
        let mut send = |downstream: &Downstream| {
            may_have_queued_scan |= downstream
                .state
                .swap(DownstreamState::Stopped)
                .may_have_queued_scan();
            let error_event = error.clone();
            if let Err(err) = downstream.sink_error_event(region_id, error_event) {
                warn!("cdc send region error failed";
                    "error" => ?err, "origin_error" => ?error, "downstream" => ?downstream.peer,
                    "downstream_id" => ?downstream.id, "request_id" => ?downstream.req_id,
                    "region_id" => region_id, "conn_id" => ?downstream.conn_id);
            } else {
                info!("cdc send region error success";
                    "origin_error" => ?error, "downstream" => ?downstream.peer,
                    "downstream_id" => ?downstream.id, "request_id" => ?downstream.req_id,
                    "region_id" => region_id, "conn_id" => ?downstream.conn_id);
            }
        };

        for downstream in &self.downstreams {
            send(downstream);
        }
        may_have_queued_scan
    }

    /// `txn_extra_op` returns a shared flag which is accessed in TiKV's
    /// transaction layer to determine whether to capture modifications' old
    /// value or not. Unsubscribing all downstreams or calling
    /// `Delegate::stop` will store it with `TxnExtraOp::Noop`.
    ///
    /// NOTE: Dropping a `Delegate` won't update this flag.
    pub fn txn_extra_op(&self) -> &AtomicCell<TxnExtraOp> {
        self.txn_extra_op.as_ref()
    }

    /// Try advance and broadcast resolved ts.
    pub(crate) fn on_min_ts(
        &mut self,
        min_ts: TimeStamp,
        current_ts: TimeStamp,
        advance: &mut Advance,
    ) {
        let locks = match &self.lock_tracker {
            LockTracker::Prepared { locks, .. } => locks,
            _ => {
                advance.blocked_on_scan += self.downstreams.len();
                let now = Instant::now_coarse();
                let elapsed = now.duration_since(self.created);
                if elapsed > WARN_LAG_THRESHOLD
                    && now.duration_since(self.last_lag_warn) > WARN_LAG_INTERVAL
                {
                    warn!(
                        "cdc region scan locks too slow";
                        "region_id" => self.region_id,
                        "elapsed" => ?elapsed,
                        "stage" => ?self.lock_tracker,
                    );
                    self.last_lag_warn = now;
                }
                return;
            }
        };

        let mut handle_downstream = |downstream: &mut Downstream| -> Option<TimeStamp> {
            if !downstream.state.load().ready_for_advancing_ts() {
                advance.blocked_on_scan += 1;
                return None;
            }
            advance.scan_finished += 1;

            if downstream.lock_heap.is_none() {
                let mut lock_heap = BTreeMap::<TimeStamp, isize>::new();
                for (_, lock) in locks.range(downstream.observed_range.to_range()) {
                    let lock_count = lock_heap.entry(lock.ts).or_default();
                    *lock_count += 1;
                }
                downstream.lock_heap = Some(lock_heap);
            }

            let lock_heap = downstream.lock_heap.as_ref().unwrap();
            let min_lock = lock_heap.keys().next().cloned().unwrap_or(min_ts);
            let advanced_to = std::cmp::min(min_lock, min_ts);
            if advanced_to > downstream.advanced_to {
                downstream.advanced_to = advanced_to;
            } else {
                advance.blocked_on_locks += 1;
            }
            Some(downstream.advanced_to)
        };

        let mut slow_downstreams = Vec::new();
        for d in &mut self.downstreams {
            let advanced_to = match handle_downstream(d) {
                Some(ts) => ts,
                None => continue,
            };

            let k = (d.conn_id, d.req_id);
            let v = advance.multiplexing.entry(k).or_default();
            v.push(self.region_id, advanced_to);

            let lag = current_ts
                .physical()
                .saturating_sub(d.advanced_to.physical());
            if Duration::from_millis(lag) > WARN_LAG_THRESHOLD {
                info!("found slow connection"; "region_id" => self.region_id,
                    "downstream_id" => ?d.id, "request_id" => ?d.req_id,
                    "conn_id" => ?d.conn_id, "lag_ms" => lag);
                slow_downstreams.push(d.id);
            }
        }

        if !slow_downstreams.is_empty() {
            let now = Instant::now_coarse();
            if now.duration_since(self.last_lag_warn) > WARN_LAG_INTERVAL {
                warn!(
                    "cdc region downstreams are too slow";
                    "region_id" => self.region_id,
                    "downstreams" => ?slow_downstreams,
                );
                self.last_lag_warn = now;
            }
        }
    }

    pub fn on_batch(
        &mut self,
        batch: CmdBatch,
        old_value_cb: &OldValueCallback,
        txn_file_loader: Option<&TxnFileLoaderContext>,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        // Stale CmdBatch, drop it silently.
        if batch.cdc_id != self.handle.id {
            return Ok(());
        }
        for cmd in batch.into_iter(self.region_id) {
            let Cmd {
                index: _,
                term: _,
                mut request,
                mut response,
                cdc_extra,
            } = cmd;
            if response.get_header().has_error() {
                let err_header = response.mut_header().take_error();
                return Err(Error::request(err_header));
            }
            if !request.has_admin_request() {
                let is_one_pc =
                    WriteBatchFlags::from_bits_truncate(request.get_header().get_flags())
                        .contains(WriteBatchFlags::ONE_PC);
                self.sink_data(
                    request.requests.into(),
                    is_one_pc,
                    old_value_cb,
                    old_value_cache,
                    statistics,
                )?;
            } else {
                self.sink_admin(request.take_admin_request(), response.take_admin_response())?;
            }
            self.sink_cdc_extra(
                cdc_extra,
                txn_file_loader,
                old_value_cb,
                old_value_cache,
                statistics,
            )?;
        }
        Ok(())
    }

    fn sink_cdc_extra(
        &mut self,
        cdc_extra: CdcExtra,
        txn_file_loader: Option<&TxnFileLoaderContext>,
        old_value_cb: &OldValueCallback,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        if let Some(txn_file) = cdc_extra.txn_file {
            self.sink_txn_file_record(
                txn_file,
                txn_file_loader,
                old_value_cb,
                old_value_cache,
                statistics,
            )?;
        }
        Ok(())
    }

    fn sink_txn_file_record(
        &mut self,
        record: TxnFileCdcRecord,
        txn_file_loader: Option<&TxnFileLoaderContext>,
        old_value_cb: &OldValueCallback,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        let loader = txn_file_loader.ok_or_else(|| {
            debug_assert!(false);
            Error::Other(box_err!(
                "missing txn-file loader context, region_id {}, region_version {}",
                self.region_id,
                record.region_version
            ))
        })?;
        if loader.region_id != self.region_id || loader.region_version != record.region_version {
            let store_err = RaftStoreError::EpochNotMatch(
                format!(
                    "stale txn-file loader context, region_id {}, loader ({}, {}), record version {}",
                    self.region_id, loader.region_id, loader.region_version, record.region_version
                ),
                vec![],
            )
            .into();
            return Err(Error::request(store_err));
        }

        let mut txn_file_refs = record.unmarshal_txn_file_refs().map_err(|err| {
            debug_assert!(false);
            Error::Other(box_err!(
                "failed to unmarshal TxnFileRefs, region_id {}, region_version {}: {:?}",
                self.region_id,
                record.region_version,
                err
            ))
        })?;
        if txn_file_refs.txn_file_refs.len() != 1 {
            debug_assert!(false);
            return Err(Error::Other(box_err!(
                "expected exactly one TxnFileRef, region_id {}, region_version {}, got {}",
                self.region_id,
                record.region_version,
                txn_file_refs.txn_file_refs.len()
            )));
        }
        let txn_file_ref = txn_file_refs.take_txn_file_refs().pop().unwrap();

        let encryption_state = loader.encryption_key.to_encryption_state();
        let txn_file = loader
            .txn_chunk_manager
            .load_txn_file_from_ref(
                self.region_id,
                record.region_version,
                &txn_file_ref,
                false,
                encryption_state,
                IoType::Export,
            )
            .map_err(|err| {
                Error::Other(box_err!(
                    "failed to load txn file, region_id {}, region_version {}, start_ts {}, chunks {:?}: {:?}",
                    self.region_id,
                    record.region_version,
                    txn_file_ref.get_start_ts(),
                    txn_file_ref.get_chunk_ids(),
                    err
                ))
            })?;
        let mut txn_file_memory = OwnedAllocated::new(self.memory_quota.clone());
        txn_file_memory.alloc(txn_file.size())?;

        let mut iter =
            SkipOpTxnFileIterator::new(TxnFileIterator::new(txn_file, false), true, true);
        iter.rewind();
        let keyspace_prefix = ApiV2::get_keyspace_prefix_by_id(loader.keyspace_id);

        if txn_file_ref.is_locked() {
            let lock = txn_types::parse_non_shared_lock(txn_file_ref.get_lock_val_prefix())
                .map_err(|err| {
                    Error::Other(box_err!(
                        "failed to parse txn-file lock prefix, region_id {}, start_ts {}: {:?}",
                        self.region_id,
                        txn_file_ref.get_start_ts(),
                        err
                    ))
                })?;
            return self.sink_locked_txn_file_rows(
                &mut iter,
                &keyspace_prefix,
                &lock,
                old_value_cb,
                old_value_cache,
                statistics,
            );
        }

        let user_meta = UserMeta::from_slice(txn_file_ref.get_user_meta());
        let is_commit = user_meta.is_commit();
        if is_commit {
            self.sink_committed_txn_file_rows(
                &mut iter,
                &keyspace_prefix,
                user_meta,
                old_value_cb,
                old_value_cache,
                statistics,
            )
        } else {
            assert!(user_meta.is_rollback());
            self.sink_rollback_txn_file_rows(
                &mut iter,
                &keyspace_prefix,
                user_meta,
                old_value_cb,
                old_value_cache,
                statistics,
            )
        }
    }

    fn sink_locked_txn_file_rows(
        &mut self,
        iter: &mut SkipOpTxnFileIterator,
        keyspace_prefix: &[u8],
        lock: &Lock,
        old_value_cb: &OldValueCallback,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        let mut rows = Vec::new();
        let mut rows_size = 0usize;

        while iter.valid() {
            let key = iter.key();
            if !is_index_key(key) {
                let mut row = RowInBuilding::default();
                let (event_key, tracking_key) =
                    build_txn_file_event_and_tracking_key(keyspace_prefix, key);
                row.v.key = event_key;
                row.v.start_ts = lock.ts.into_inner();
                row.v.op_type = match lock.lock_type {
                    LockType::Put => EventRowOpType::Put,
                    LockType::Delete => EventRowOpType::Delete,
                    _ => EventRowOpType::Unknown,
                } as _;
                row.v.txn_source = lock.txn_source;
                set_event_row_type(&mut row.v, EventLogType::Prewrite);
                row.lock_count_modify =
                    self.push_lock(tracking_key, MiniLock::new(lock.ts, lock.txn_source))?;
                rows_size += row.v.key.len();
                rows.push(row);
            }

            if rows_size >= CDC_EVENT_MAX_BYTES {
                self.flush_txn_file_rows(&mut rows, old_value_cb, old_value_cache, statistics)?;
                rows_size = 0;
            }
            iter.next();
        }

        self.flush_txn_file_rows(&mut rows, old_value_cb, old_value_cache, statistics)
    }

    fn sink_committed_txn_file_rows(
        &mut self,
        iter: &mut SkipOpTxnFileIterator,
        keyspace_prefix: &[u8],
        user_meta: UserMeta,
        old_value_cb: &OldValueCallback,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        let mut rows = Vec::new();
        let mut rows_size = 0usize;

        while iter.valid() {
            let key = iter.key();
            if !is_index_key(key) {
                let value = iter.value();
                if value.version != user_meta.commit_ts {
                    debug_assert!(false);
                    return Err(Error::Other(box_err!(
                        "txn-file commit_ts mismatch, region_id {}, start_ts {}, user_meta {}, value {}",
                        self.region_id,
                        user_meta.start_ts,
                        user_meta.commit_ts,
                        value.version
                    )));
                }
                let mut row = RowInBuilding::default();
                let (event_key, tracking_key) =
                    build_txn_file_event_and_tracking_key(keyspace_prefix, key);
                row.v.key = event_key;
                row.v.value = value.get_value().to_vec();
                row.v.start_ts = user_meta.start_ts;
                row.v.commit_ts = value.version;
                row.v.op_type = if row.v.value.is_empty() {
                    EventRowOpType::Delete
                } else {
                    EventRowOpType::Put
                } as _;
                // TODO: Support txn_source.
                set_event_row_type(&mut row.v, EventLogType::Committed);
                row.needs_old_value = Some(TimeStamp::from(value.version).prev());
                row.lock_count_modify =
                    self.pop_lock(tracking_key, TimeStamp::from(user_meta.start_ts))?;
                rows_size += row.v.key.len() + row.v.value.len();
                rows.push(row);
            }

            if rows_size >= CDC_EVENT_MAX_BYTES {
                self.flush_txn_file_rows(&mut rows, old_value_cb, old_value_cache, statistics)?;
                rows_size = 0;
            }
            iter.next();
        }

        self.flush_txn_file_rows(&mut rows, old_value_cb, old_value_cache, statistics)
    }

    fn sink_rollback_txn_file_rows(
        &mut self,
        iter: &mut SkipOpTxnFileIterator,
        keyspace_prefix: &[u8],
        user_meta: UserMeta,
        old_value_cb: &OldValueCallback,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        let mut rows = Vec::new();
        let mut rows_size = 0usize;

        while iter.valid() {
            let key = iter.key();
            if !is_index_key(key) {
                let mut row = RowInBuilding::default();
                let (event_key, tracking_key) =
                    build_txn_file_event_and_tracking_key(keyspace_prefix, key);
                row.v.key = event_key;
                row.v.start_ts = user_meta.start_ts;
                // Leave the row type as Unknown: txn-file rollback rows only
                // update downstream lock heaps. `sink_downstream_tidb` applies
                // `lock_count_modify` before filtering Unknown rows from the
                // downstream event stream.
                row.lock_count_modify =
                    self.pop_lock(tracking_key, TimeStamp::from(user_meta.start_ts))?;
                rows_size += row.v.key.len();
                rows.push(row);
            }

            if rows_size >= CDC_EVENT_MAX_BYTES {
                self.flush_txn_file_rows(&mut rows, old_value_cb, old_value_cache, statistics)?;
                rows_size = 0;
            }
            iter.next();
        }

        self.flush_txn_file_rows(&mut rows, old_value_cb, old_value_cache, statistics)
    }

    fn flush_txn_file_rows(
        &mut self,
        rows: &mut Vec<RowInBuilding>,
        old_value_cb: &OldValueCallback,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        if rows.is_empty() {
            return Ok(());
        }
        let entries = std::mem::replace(rows, Vec::with_capacity(rows.len()));
        self.sink_downstream_tidb(entries, |row, read_old_ts| {
            let key = Key::from_raw(&row.key).append_ts(row.start_ts.into());
            let old_value = old_value_cb(key, read_old_ts, old_value_cache, statistics)?;
            row.old_value = old_value.unwrap_or_default();
            Ok(())
        })
    }

    pub(crate) fn convert_to_grpc_events(
        region_id: u64,
        request_id: RequestId,
        entries: Vec<Option<TxnEntry>>,
        filter_loop: bool,
        observed_range: &ObservedRange,
    ) -> Result<Vec<CdcEvent>> {
        let entries_len = entries.len();
        let mut rows = vec![Vec::with_capacity(entries_len)];
        let mut current_rows_size: usize = 0;
        for entry in entries {
            let (mut row, mut _has_value) = (EventRow::default(), false);
            let row_size: usize;
            match entry {
                Some(TxnEntry::Prewrite {
                    default,
                    lock,
                    old_value,
                }) => {
                    if !observed_range.contains_encoded_key(&lock.0) {
                        continue;
                    }
                    let lock_type = txn_types::decode_lock_type(&lock.1).unwrap();
                    if !matches!(lock_type, LockType::Put | LockType::Delete) {
                        // CDC only emits prewrite rows for locks that may
                        // carry a user-visible data change. Plain `Lock`,
                        // pessimistic locks, and shared-lock wrappers are all
                        // lock-mode metadata on the lock CF path.
                        //
                        // Intentionally do not distinguish shared pessimistic
                        // holders from shared prewrite holders here: neither
                        // stage contains row data, and shared-lock
                        // transactions later publish at most a write-CF
                        // `WriteType::Lock` marker rather than a downstream row
                        // mutation.
                        debug!(
                            "cdc skip lock record";
                            "lock" => ?lock_type,
                            "key" => log_wrappers::Value::key(&lock.0),
                        );
                        continue;
                    }
                    let l = txn_types::parse_lock(&lock.1)
                        .unwrap()
                        .left()
                        .expect("only put/delete lock should be here");
                    if decode_lock(lock.0, l, &mut row, &mut _has_value) {
                        continue;
                    }
                    decode_default(default.1, &mut row, &mut _has_value);
                    row.old_value = old_value.finalized().unwrap_or_default();
                    row_size = row.key.len() + row.value.len() + row.old_value.len();
                }
                Some(TxnEntry::Commit {
                    default,
                    write,
                    old_value,
                }) => {
                    if !observed_range.contains_encoded_key(&write.0) {
                        continue;
                    }
                    if decode_write(write.0, &write.1, &mut row, &mut _has_value, false) {
                        continue;
                    }
                    decode_default(default.1, &mut row, &mut _has_value);

                    // This type means the row is self-contained, it has,
                    //   1. start_ts
                    //   2. commit_ts
                    //   3. key
                    //   4. value
                    if row.get_type() == EventLogType::Rollback {
                        // We dont need to send rollbacks to downstream,
                        // because downstream does not needs rollback to clean
                        // prewrite as it drops all previous stashed data.
                        continue;
                    }
                    set_event_row_type(&mut row, EventLogType::Committed);
                    row.old_value = old_value.finalized().unwrap_or_default();
                    row_size = row.key.len() + row.value.len() + row.old_value.len();
                }
                None => {
                    // This type means scan has finished.
                    set_event_row_type(&mut row, EventLogType::Initialized);
                    row_size = 0;
                }
            }
            if TxnSource::is_lightning_physical_import(row.txn_source)
                || TxnSource::is_lossy_ddl_reorg_source_set(row.txn_source)
                || filter_loop && TxnSource::is_cdc_write_source_set(row.txn_source)
                // In next-gen, all events become `Committed` as we can always observe the value when committing.
                // Hence `Prewrite` entries is nolonger needed by the client.
                || row.get_type() == EventLogType::Prewrite || row.get_type() == EventLogType::Unknown
            {
                continue;
            }
            if current_rows_size + row_size >= CDC_EVENT_MAX_BYTES {
                rows.push(Vec::with_capacity(entries_len));
                current_rows_size = 0;
            }
            current_rows_size += row_size;
            rows.last_mut().unwrap().push(row);
        }

        let rows = rows
            .into_iter()
            .filter(|rs| !rs.is_empty())
            .map(|rs| {
                let event_entries = EventEntries {
                    entries: rs.into(),
                    ..Default::default()
                };
                CdcEvent::Event(Event {
                    region_id,
                    request_id: request_id.0,
                    event: Some(Event_oneof_event::Entries(event_entries)),
                    ..Default::default()
                })
            })
            .collect();
        Ok(rows)
    }

    fn sink_data(
        &mut self,
        requests: Vec<Request>,
        is_one_pc: bool,
        old_value_cb: &OldValueCallback,
        old_value_cache: &mut OldValueCache,
        statistics: &mut Statistics,
    ) -> Result<()> {
        debug_assert_eq!(self.txn_extra_op.load(), TxnExtraOp::ReadOldValue);

        let read_old_value = |row: &mut EventRow, read_old_ts| -> Result<()> {
            let key = Key::from_raw(&row.key).append_ts(row.start_ts.into());
            let old_value = old_value_cb(key, read_old_ts, old_value_cache, statistics)?;
            row.old_value = old_value.unwrap_or_default();
            Ok(())
        };

        let mut rows_builder = RowsBuilder::default();
        rows_builder.is_one_pc = is_one_pc;
        for mut req in requests {
            match req.get_cmd_type() {
                CmdType::Put => self.sink_txn_put(req.take_put(), &mut rows_builder)?,
                _ => debug!("cdc skip other command";
                    "region_id" => self.region_id,
                    "command" => ?req),
            };
        }

        let txns = rows_builder.finish_build();
        self.sink_downstream_tidb(txns, read_old_value)?;
        Ok(())
    }

    fn sink_downstream_tidb(
        &mut self,
        mut entries: Vec<RowInBuilding>,
        mut read_old_value: impl FnMut(&mut EventRow, TimeStamp) -> Result<()>,
    ) -> Result<()> {
        let mut downstreams = Vec::with_capacity(self.downstreams.len());
        for d in &mut self.downstreams {
            if d.state.load().ready_for_change_events() {
                downstreams.push(d);
            }
        }
        if downstreams.is_empty() {
            return Ok(());
        }

        let entries_len = entries.len();
        for downstream in downstreams {
            let mut filtered_entries = Vec::with_capacity(entries_len);
            let mut filtered_entries_size = 0usize;
            for RowInBuilding {
                v,
                lock_count_modify,
                needs_old_value,
                ..
            } in &mut entries
            {
                if !downstream.observed_range.contains_raw_key(&v.key) {
                    continue;
                }

                if let Some(read_old_ts) = needs_old_value {
                    read_old_value(v, *read_old_ts)?;
                    *needs_old_value = None;
                }

                if *lock_count_modify != 0 {
                    if let Some(lock_heap) = downstream.lock_heap.as_mut() {
                        match lock_heap.entry(v.start_ts.into()) {
                            BTreeMapEntry::Vacant(x) => {
                                x.insert(*lock_count_modify);
                            }
                            BTreeMapEntry::Occupied(mut x) => {
                                *x.get_mut() += *lock_count_modify;
                                assert!(
                                    *x.get() >= 0,
                                    "lock_count_modify should never be negative, start_ts: {}",
                                    v.start_ts
                                );
                                if *x.get() == 0 {
                                    x.remove();
                                }
                            }
                        }
                    }
                }

                if TxnSource::is_lightning_physical_import(v.txn_source)
                    || TxnSource::is_lossy_ddl_reorg_source_set(v.txn_source)
                    || downstream.filter_loop && TxnSource::is_cdc_write_source_set(v.txn_source)
                    // In next-gen, all events become `Committed` as we can always observe the value when committing.
                    || v.get_type() == EventLogType::Prewrite || v.get_type() == EventLogType::Unknown
                {
                    continue;
                }

                let row_size = v.key.len() + v.value.len() + v.old_value.len();
                if filtered_entries_size + row_size >= CDC_EVENT_MAX_BYTES
                    && !filtered_entries.is_empty()
                {
                    let event = Event {
                        region_id: self.region_id,
                        request_id: downstream.req_id.0,
                        event: Some(Event_oneof_event::Entries(EventEntries {
                            entries: std::mem::replace(
                                &mut filtered_entries,
                                Vec::with_capacity(entries_len),
                            )
                            .into(),
                            ..Default::default()
                        })),
                        ..Default::default()
                    };
                    downstream.sink_event(event, false)?;
                    filtered_entries_size = 0;
                }
                filtered_entries_size += row_size;
                filtered_entries.push(v.clone());
            }
            if filtered_entries.is_empty() {
                continue;
            }
            let event = Event {
                region_id: self.region_id,
                request_id: downstream.req_id.0,
                event: Some(Event_oneof_event::Entries(EventEntries {
                    entries: filtered_entries.into(),
                    ..Default::default()
                })),
                ..Default::default()
            };
            downstream.sink_event(event, false)?;
        }
        Ok(())
    }

    fn sink_txn_put(&mut self, mut put: PutRequest, rows: &mut RowsBuilder) -> Result<()> {
        match put.cf.as_str() {
            "write" => {
                let key = Key::from_encoded_slice(&put.key).truncate_ts().unwrap();
                let row = rows.txns_by_key.entry(key.clone()).or_default();
                if decode_write(
                    put.take_key(),
                    &put.value,
                    &mut row.v,
                    &mut row.has_value,
                    true,
                ) {
                    return Ok(());
                }

                let start_ts = TimeStamp::from(row.v.start_ts);
                if row.v.r_type == EventLogType::Commit {
                    let read_old_ts = TimeStamp::from(row.v.commit_ts).prev();
                    row.needs_old_value = Some(read_old_ts);
                    assert_eq!(row.lock_count_modify, 0);
                    // In next-gen, here must be a default cf modification or short value.
                    set_event_row_type(&mut row.v, EventLogType::Committed);
                }
                row.lock_count_modify = self.pop_lock(key, start_ts)?;
            }
            "lock" => {
                let lock_type = txn_types::decode_lock_type(put.get_value()).unwrap();
                if !matches!(lock_type, LockType::Put | LockType::Delete) {
                    // CDC's lock heap only tracks prewrite locks that can lead
                    // to downstream row changes. Plain `Lock`, pessimistic
                    // locks, and shared-lock wrappers are intentionally
                    // ignored as a whole.
                    //
                    // The shared holder stage does not matter here: both
                    // shared pessimistic and shared prewrite states are pure
                    // lock-mode metadata and do not need old-value or lock-heap
                    // tracking on the CDC apply path.
                    debug!(
                        "cdc skip lock record";
                        "lock" => ?lock_type,
                        "key" => log_wrappers::Value::key(put.get_key()),
                    );
                    return Ok(());
                }
                let lock = txn_types::parse_lock(put.get_value())
                    .unwrap()
                    .left()
                    .expect("only put/delete lock should be here");
                let txn_source = lock.txn_source;
                // let for_update_ts = lock.for_update_ts;
                let key = Key::from_encoded_slice(&put.key);
                let row = rows.txns_by_key.entry(key.clone()).or_default();
                if decode_lock(put.take_key(), lock, &mut row.v, &mut row.has_value) {
                    return Ok(());
                }

                assert_eq!(row.lock_count_modify, 0);
                let mini_lock = MiniLock::new(row.v.start_ts, txn_source);
                row.lock_count_modify = self.push_lock(key, mini_lock)?;
                // No need to fetch old_value in next-gen.
                // let read_old_ts = std::cmp::max(for_update_ts,
                // row.v.start_ts.into()); row.needs_old_value =
                // Some(read_old_ts);
            }
            "" | "default" => {
                let key = Key::from_encoded(put.take_key()).truncate_ts().unwrap();
                let row = rows.txns_by_key.entry(key).or_default();
                decode_default(put.take_value(), &mut row.v, &mut row.has_value);
            }
            other => panic!("invalid cf {}", other),
        }
        Ok(())
    }

    fn sink_admin(&mut self, request: AdminRequest, mut response: AdminResponse) -> Result<()> {
        let store_err = match request.get_cmd_type() {
            AdminCmdType::Split => RaftStoreError::EpochNotMatch(
                "split".to_owned(),
                vec![
                    response.mut_split().take_left(),
                    response.mut_split().take_right(),
                ],
            ),
            AdminCmdType::BatchSplit => RaftStoreError::EpochNotMatch(
                "batchsplit".to_owned(),
                response.mut_splits().take_regions().into(),
            ),
            AdminCmdType::PrepareMerge
            | AdminCmdType::CommitMerge
            | AdminCmdType::RollbackMerge => {
                RaftStoreError::EpochNotMatch("merge".to_owned(), vec![])
            }
            _ => return Ok(()),
        };
        Err(Error::request(store_err.into()))
    }

    fn add_downstream(&mut self, downstream: Downstream) {
        self.downstreams_mut().push(downstream);
        self.txn_extra_op.store(TxnExtraOp::ReadOldValue);
    }

    fn remove_downstream(&mut self, id: DownstreamId) -> Option<Downstream> {
        if let Some(index) = self.downstreams.iter().position(|x| x.id == id) {
            let downstream = self.downstreams.swap_remove(index);
            if self.downstreams.is_empty() {
                // Stop observing when the last downstream is removed. Otherwise the observer
                // will keep pushing events to the delegate.
                self.stop_observing();
            }
            return Some(downstream);
        }
        None
    }

    fn check_epoch_on_ready(downstream: &Downstream, region: &Region) -> Result<()> {
        if let Err(e) = compare_region_epoch(
            &downstream.region_epoch,
            region,
            false, // check_conf_ver
            true,  // check_ver
            true,  // include_region
        ) {
            info!(
                "cdc fail to subscribe downstream";
                "region_id" => region.id,
                "downstream_id" => ?downstream.id,
                "conn_id" => ?downstream.conn_id,
                "req_id" => ?downstream.req_id,
                "err" => ?e
            );
            // Downstream is outdated, mark stop.
            downstream.state.store(DownstreamState::Stopped);
            return Err(Error::request(e.into()));
        }
        Ok(())
    }

    fn stop_observing(&self) {
        info!("cdc stop observing"; "failed" => self.failed, "region_id" => self.region_id);
        // Stop observe further events.
        self.handle.stop_observing();
        // To inform transaction layer no more old values are required for the region.
        self.txn_extra_op.store(TxnExtraOp::Noop);
    }
}

#[derive(Default)]
struct RowsBuilder {
    txns_by_key: HashMap<Key, RowInBuilding>,
    is_one_pc: bool,
}

#[derive(Default, Debug)]
struct RowInBuilding {
    v: EventRow,
    has_value: bool,
    lock_count_modify: isize,
    needs_old_value: Option<TimeStamp>,
}

impl RowsBuilder {
    fn finish_build(self) -> Vec<RowInBuilding> {
        let mut txns = Vec::with_capacity(self.txns_by_key.len());
        for row in self.txns_by_key.into_values() {
            if row.v.r_type == EventLogType::Prewrite
                && row.v.op_type == EventRowOpType::Put
                && !row.has_value
            {
                // It's possible that a prewrite command only contains lock but without
                // default. It's not documented by classic Percolator but introduced with
                // Large-Transaction. Those prewrites are not complete, we must skip them.
                continue;
            }
            // defensive assertion for next-gen TiKV: where `Prewrite` and `Commit` are no
            // longer used and only `Committed`s are sent.
            assert!(
                !(row.v.r_type == EventLogType::Committed
                    && row.v.op_type == EventRowOpType::Put
                    && row.v.value.is_empty()),
                "committed put event without value, this can be a bug; row = {row:?}"
            );
            txns.push(row);
        }
        txns
    }
}

fn set_event_row_type(row: &mut EventRow, ty: EventLogType) {
    row.r_type = ty;
}

fn make_overlapped_rollback(key: Key, row: &mut EventRow) {
    // The current record's commit_ts is the rolled-back transaction's start_ts.
    row.start_ts = key.decode_ts().unwrap().into_inner();
    row.commit_ts = 0;
    row.key = key.truncate_ts().unwrap().into_raw().unwrap();
    row.op_type = EventRowOpType::Unknown as _;
    set_event_row_type(row, EventLogType::Rollback);
}

/// Decodes the write record and store its information in `row`. This may be
/// called both when doing incremental scan of observing apply events. There's
/// different behavior for the two case, distinguished by the `is_apply`
/// parameter.
fn decode_write(
    key: Vec<u8>,
    value: &[u8],
    row: &mut EventRow,
    has_value: &mut bool,
    is_apply: bool,
) -> bool {
    let key = Key::from_encoded(key);
    let write = WriteRef::parse(value).unwrap().to_owned();

    // For scanning, ignore the GC fence and read the old data;
    // For observed apply, drop the record it self but keep only the overlapped
    // rollback information if gc_fence exists.
    if is_apply && write.gc_fence.is_some() {
        // `gc_fence` is set means the write record has been rewritten.
        // Currently the only case is writing overlapped_rollback. And in this case
        assert!(write.has_overlapped_rollback);
        assert_ne!(write.write_type, WriteType::Rollback);
        make_overlapped_rollback(key, row);
        return false;
    }

    let (op_type, r_type) = match write.write_type {
        WriteType::Put => (EventRowOpType::Put, EventLogType::Commit),
        WriteType::Delete => (EventRowOpType::Delete, EventLogType::Commit),
        WriteType::Rollback => (EventRowOpType::Unknown, EventLogType::Rollback),
        other => {
            debug!("cdc skip write record"; "write" => ?other, "key" => %key);
            return true;
        }
    };
    let commit_ts = if write.write_type == WriteType::Rollback {
        assert_eq!(write.txn_source, 0);
        0
    } else {
        key.decode_ts().unwrap().into_inner()
    };
    row.start_ts = write.start_ts.into_inner();
    row.commit_ts = commit_ts;
    row.key = key.truncate_ts().unwrap().into_raw().unwrap();
    row.op_type = op_type as _;
    // used for filter out the event. see `txn_source` field for more detail.
    row.txn_source = write.txn_source;
    set_event_row_type(row, r_type);
    if let Some(value) = write.short_value {
        row.value = value;
        *has_value = true;
    }

    false
}

fn decode_lock(key: Vec<u8>, mut lock: Lock, row: &mut EventRow, has_value: &mut bool) -> bool {
    let key = Key::from_encoded(key);
    let op_type = match lock.lock_type {
        LockType::Put => EventRowOpType::Put,
        LockType::Delete => EventRowOpType::Delete,
        other => {
            debug!("cdc skip lock record"; "lock" => ?other, "key" => %key);
            return true;
        }
    };

    // We need to fill some metadata used internally in cse-cdc.
    row.start_ts = lock.ts.into_inner();
    row.key = key.into_raw().unwrap();
    row.op_type = op_type as _;
    row.txn_source = lock.txn_source;
    set_event_row_type(row, EventLogType::Prewrite);
    if let Some(value) = lock.short_value.take() {
        assert!(!*has_value, "unexpected lock with value: {:?}", lock);
        *has_value = true;
        row.value = value;
    }

    false
}

fn decode_default(value: Vec<u8>, row: &mut EventRow, has_value: &mut bool) {
    if !value.is_empty() {
        row.value = value.to_vec();
    }
    // If default CF is given in a command it means the command always has a value.
    *has_value = true;
}

#[inline]
fn is_index_key(key: InnerKey<'_>) -> bool {
    tidb_query_datatype::codec::table::is_index_key(&key)
}

#[inline]
fn build_txn_file_event_key(keyspace_prefix: &[u8], inner_key: InnerKey<'_>) -> Vec<u8> {
    if keyspace_prefix.is_empty() {
        return inner_key.to_vec();
    }
    let mut outer_key = Vec::with_capacity(keyspace_prefix.len() + inner_key.len());
    outer_key.extend_from_slice(keyspace_prefix);
    outer_key.extend_from_slice(&inner_key);
    outer_key
}

#[inline]
fn build_txn_file_event_and_tracking_key(
    keyspace_prefix: &[u8],
    inner_key: InnerKey<'_>,
) -> (Vec<u8>, Key) {
    let event_key = build_txn_file_event_key(keyspace_prefix, inner_key);
    let tracking_key = Key::from_raw(&event_key);
    (event_key, tracking_key)
}

/// Observed key range.
#[derive(Clone)]
pub struct ObservedRange {
    pub start_key_encoded: Key,
    pub end_key_encoded: Key,
    pub start_key_raw: Vec<u8>,
    pub end_key_raw: Vec<u8>,
    pub all_key_covered: bool,
}

impl Default for ObservedRange {
    fn default() -> Self {
        ObservedRange {
            start_key_encoded: Key::from_encoded(vec![]),
            end_key_encoded: Key::from_encoded(vec![]),
            start_key_raw: vec![],
            end_key_raw: vec![],
            all_key_covered: false,
        }
    }
}

impl ObservedRange {
    pub fn new(start_key_encoded: Vec<u8>, end_key_encoded: Vec<u8>) -> Result<ObservedRange> {
        let start_key_encoded = Key::from_encoded(start_key_encoded);
        let end_key_encoded = Key::from_encoded(end_key_encoded);
        let start_key_raw = start_key_encoded
            .clone()
            .into_raw()
            .map_err(|e| Error::Other(e.into()))?;
        let end_key_raw = end_key_encoded
            .clone()
            .into_raw()
            .map_err(|e| Error::Other(e.into()))?;
        Ok(ObservedRange {
            start_key_encoded,
            end_key_encoded,
            start_key_raw,
            end_key_raw,
            all_key_covered: false,
        })
    }

    #[allow(clippy::collapsible_if)]
    pub fn update_region_key_range(&mut self, region: &Region) {
        // Check observed key range in region.
        if self.start_key_encoded.as_encoded() <= &region.start_key {
            if self.end_key_encoded.is_empty()
                || (&region.end_key <= self.end_key_encoded.as_encoded()
                    && !region.end_key.is_empty())
            {
                // Observed range covers the region.
                self.all_key_covered = true;
            }
        }
    }

    fn is_key_in_range(&self, start_key: &[u8], end_key: &[u8], key: &[u8]) -> bool {
        if self.all_key_covered {
            return true;
        }
        if start_key <= key && (key < end_key || end_key.is_empty()) {
            return true;
        }
        false
    }

    pub fn contains_encoded_key(&self, key: &[u8]) -> bool {
        self.is_key_in_range(
            self.start_key_encoded.as_encoded(),
            self.end_key_encoded.as_encoded(),
            key,
        )
    }

    pub fn contains_raw_key(&self, key: &[u8]) -> bool {
        self.is_key_in_range(&self.start_key_raw, &self.end_key_raw, key)
    }

    pub fn filter_entries(&self, mut entries: Vec<EventRow>) -> Vec<EventRow> {
        if self.all_key_covered {
            return entries;
        }
        // Entry's key is in raw key format.
        entries.retain(|e| self.is_key_in_range(&self.start_key_raw, &self.end_key_raw, &e.key));
        entries
    }

    fn to_range(&self) -> (Bound<&Key>, Bound<&Key>) {
        let start = Bound::Included(&self.start_key_encoded);
        let end = if self.end_key_encoded.is_empty() {
            Bound::Unbounded
        } else {
            Bound::Excluded(&self.end_key_encoded)
        };
        (start, end)
    }
}

const WARN_LAG_THRESHOLD: Duration = Duration::from_secs(600);
const WARN_LAG_INTERVAL: Duration = Duration::from_secs(60);

#[cfg(test)]
mod tests {
    use std::cell::Cell;

    use api_version::ApiV2;
    use futures::{executor::block_on, stream::StreamExt};
    use kvengine::{
        table::InnerKey,
        test_engine::{DEF_BLOCK_SIZE, TABLE_KEY_PREFIX, new_test_engine_opt},
    };
    use kvproto::{errorpb::Error as ErrorHeader, metapb::Region};
    use tikv_util::{config::ReadableSize, memory::MemoryQuota};

    use super::*;
    use crate::channel::{channel, recv_timeout};

    fn shared_lock_value(start_ts: u64) -> Vec<u8> {
        let mut shared_locks = txn_types::SharedLocks::new();
        shared_locks
            .insert_lock(Lock::new(
                LockType::Lock,
                b"primary".to_vec(),
                start_ts.into(),
                3000,
                None,
                TimeStamp::zero(),
                1,
                TimeStamp::zero(),
            ))
            .unwrap();
        shared_locks.to_bytes()
    }

    #[test]
    fn test_txn_file_tracking_key_uses_outer_event_key() {
        let empty_prefix_inner_key = InnerKey::from_inner_buf(b"t123_r456");
        let (event_key, tracking_key) =
            build_txn_file_event_and_tracking_key(&[], empty_prefix_inner_key);
        assert_eq!(event_key, empty_prefix_inner_key.to_vec());
        assert_eq!(tracking_key.to_raw().unwrap(), event_key);

        let keyspace_prefix = ApiV2::get_txn_keyspace_prefix(1);
        let inner_key = InnerKey::from_inner_buf(b"t123_r456");
        let (event_key, tracking_key) =
            build_txn_file_event_and_tracking_key(&keyspace_prefix, inner_key);
        let observed_range = ObservedRange::new(
            Key::from_raw(&keyspace_prefix).into_encoded(),
            Key::from_raw(&ApiV2::get_txn_keyspace_prefix(2)).into_encoded(),
        )
        .unwrap();

        assert!(event_key.starts_with(&keyspace_prefix));
        assert_eq!(tracking_key.to_raw().unwrap(), event_key);
        assert!(observed_range.contains_raw_key(&event_key));

        let old_inner_tracking_key = Key::from_raw(&inner_key);
        assert!(!observed_range.contains_raw_key(&old_inner_tracking_key.to_raw().unwrap()));

        let mut locks = BTreeMap::new();
        locks.insert(tracking_key, MiniLock::from_ts(TimeStamp::new(1)));
        assert_eq!(locks.range(observed_range.to_range()).count(), 1);

        locks.clear();
        locks.insert(old_inner_tracking_key, MiniLock::from_ts(TimeStamp::new(1)));
        assert_eq!(locks.range(observed_range.to_range()).count(), 0);
    }

    #[test]
    fn test_txn_file_cdc_loader_version_mismatch_returns_epoch_not_match() {
        let (engine, _tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, TABLE_KEY_PREFIX);
        let mut delegate = Delegate::new(
            1,
            Arc::new(MemoryQuota::new(usize::MAX)),
            Default::default(),
        );
        let loader = TxnFileLoaderContext {
            region_id: 1,
            region_version: 8,
            keyspace_id: 1,
            txn_chunk_manager: engine.get_txn_chunk_manager(),
            encryption_key: None,
        };
        let record = TxnFileCdcRecord::new(7, b"unused-before-version-check".to_vec().into());
        let old_value_cb: OldValueCallback = Box::new(|_, _, _, _| Ok(None));
        let mut old_value_cache = OldValueCache::new(ReadableSize(usize::MAX as u64));
        let mut statistics = Statistics::default();

        let err = delegate
            .sink_txn_file_record(
                record,
                Some(&loader),
                &old_value_cb,
                &mut old_value_cache,
                &mut statistics,
            )
            .unwrap_err();

        match err {
            Error::Request(err) => assert!(err.has_epoch_not_match(), "{:?}", err),
            other => panic!("unexpected error: {:?}", other),
        }
    }

    #[test]
    fn test_convert_to_grpc_events_skips_shared_lock_prewrite() {
        let key = Key::from_raw(b"k1").into_encoded();
        let events = Delegate::convert_to_grpc_events(
            1,
            RequestId(1),
            vec![Some(TxnEntry::Prewrite {
                default: (vec![], vec![]),
                lock: (key, shared_lock_value(10)),
                old_value: txn_types::OldValue::None,
            })],
            false,
            &ObservedRange::default(),
        )
        .unwrap();

        assert!(events.is_empty());
    }

    #[test]
    fn test_sink_txn_put_skips_shared_lock_prewrite() {
        let quota = Arc::new(MemoryQuota::new(usize::MAX));
        let mut delegate = Delegate::new(1, quota, Default::default());
        let mut rows = RowsBuilder::default();
        let mut put = PutRequest::default();
        put.set_cf("lock".to_owned());
        put.set_key(Key::from_raw(b"k1").into_encoded());
        put.set_value(shared_lock_value(10));

        delegate.sink_txn_put(put, &mut rows).unwrap();

        assert!(rows.txns_by_key.is_empty());
    }

    #[test]
    fn test_error() {
        let region_id = 1;
        let mut region = Region::default();
        region.set_id(region_id);
        region.mut_peers().push(Default::default());
        region.mut_region_epoch().set_version(2);
        region.mut_region_epoch().set_conf_ver(2);
        let region_epoch = region.get_region_epoch().clone();

        let quota = Arc::new(MemoryQuota::new(usize::MAX));
        let (sink, mut drain) = channel(ConnId::default(), 1, quota.clone());
        let rx = drain.drain();
        let request_id = RequestId(123);
        let mut downstream = Downstream::new(
            String::new(),
            region_epoch,
            request_id,
            ConnId::new(),
            false,
            ObservedRange::default(),
        );
        downstream.set_sink(sink);

        let mut delegate = Delegate::new(region_id, quota, Default::default());
        delegate.subscribe(downstream).unwrap();
        assert!(delegate.handle.is_observing());

        assert!(delegate.init_lock_tracker());
        let fails = delegate
            .finish_scan_locks(region, Default::default())
            .unwrap();
        assert!(fails.is_empty());
        assert!(delegate.downstreams[0].observed_range.all_key_covered);

        let rx_wrap = Cell::new(Some(rx));
        let receive_error = || {
            let (event, rx) = block_on(rx_wrap.replace(None).unwrap().into_future());
            rx_wrap.set(Some(rx));
            if let CdcEvent::Event(mut e) = event.unwrap().0 {
                assert_eq!(e.get_request_id(), request_id.0);
                let event = e.event.take().unwrap();
                match event {
                    Event_oneof_event::Error(err) => err,
                    other => panic!("unknown event {:?}", other),
                }
            } else {
                panic!("unknown event")
            }
        };

        let mut err_header = ErrorHeader::default();
        err_header.set_not_leader(Default::default());
        assert!(delegate.stop_with_scan_cleanup(Error::request(err_header)));
        let err = receive_error();
        assert!(err.has_not_leader());
        // Observing is disabled by any error.
        assert!(!delegate.handle.is_observing());

        let mut err_header = ErrorHeader::default();
        err_header.set_region_not_found(Default::default());
        delegate.stop(Error::request(err_header));
        let err = receive_error();
        assert!(err.has_region_not_found());

        delegate.stop(Error::Sink(SendError::Congested));
        let err = receive_error();
        assert!(err.has_congested());

        let mut err_header = ErrorHeader::default();
        err_header.set_epoch_not_match(Default::default());
        delegate.stop(Error::request(err_header));
        let err = receive_error();
        assert!(err.has_epoch_not_match());

        // Split
        let mut region = Region::default();
        region.set_id(1);
        let mut request = AdminRequest::default();
        request.set_cmd_type(AdminCmdType::Split);
        let mut response = AdminResponse::default();
        response.mut_split().set_left(region.clone());
        let err = delegate.sink_admin(request, response).err().unwrap();
        delegate.stop(err);
        let mut err = receive_error();
        assert!(err.has_epoch_not_match());
        err.take_epoch_not_match()
            .current_regions
            .into_iter()
            .find(|r| r.get_id() == 1)
            .unwrap();

        let mut request = AdminRequest::default();
        request.set_cmd_type(AdminCmdType::BatchSplit);
        let mut response = AdminResponse::default();
        response.mut_splits().set_regions(vec![region].into());
        let err = delegate.sink_admin(request, response).err().unwrap();
        delegate.stop(err);
        let mut err = receive_error();
        assert!(err.has_epoch_not_match());
        err.take_epoch_not_match()
            .current_regions
            .into_iter()
            .find(|r| r.get_id() == 1)
            .unwrap();

        // Merge
        let mut request = AdminRequest::default();
        request.set_cmd_type(AdminCmdType::PrepareMerge);
        let response = AdminResponse::default();
        let err = delegate.sink_admin(request, response).err().unwrap();
        delegate.stop(err);
        let mut err = receive_error();
        assert!(err.has_epoch_not_match());
        assert!(err.take_epoch_not_match().current_regions.is_empty());

        let mut request = AdminRequest::default();
        request.set_cmd_type(AdminCmdType::CommitMerge);
        let response = AdminResponse::default();
        let err = delegate.sink_admin(request, response).err().unwrap();
        delegate.stop(err);
        let mut err = receive_error();
        assert!(err.has_epoch_not_match());
        assert!(err.take_epoch_not_match().current_regions.is_empty());

        let mut request = AdminRequest::default();
        request.set_cmd_type(AdminCmdType::RollbackMerge);
        let response = AdminResponse::default();
        let err = delegate.sink_admin(request, response).err().unwrap();
        delegate.stop(err);
        let mut err = receive_error();
        assert!(err.has_epoch_not_match());
        assert!(err.take_epoch_not_match().current_regions.is_empty());
    }

    #[test]
    fn test_delegate_subscribe_unsubscribe() {
        let new_downstream = |id: RequestId, region_version: u64| {
            let peer = format!("{:?}", id);
            let mut epoch = RegionEpoch::default();
            epoch.set_conf_ver(region_version);
            epoch.set_version(region_version);
            Downstream::new(
                peer,
                epoch,
                id,
                ConnId::new(),
                false,
                ObservedRange::default(),
            )
        };

        // Create a new delegate.
        let memory_quota = Arc::new(MemoryQuota::new(usize::MAX));
        let txn_extra_op = Arc::new(AtomicCell::new(TxnExtraOp::Noop));
        let mut delegate = Delegate::new(1, memory_quota, txn_extra_op.clone());
        assert_eq!(txn_extra_op.load(), TxnExtraOp::Noop);
        assert!(delegate.handle.is_observing());

        // Subscribe once.
        let downstream1 = new_downstream(RequestId(1), 1);
        let downstream1_id = downstream1.id;
        delegate.subscribe(downstream1).unwrap();
        assert_eq!(txn_extra_op.load(), TxnExtraOp::ReadOldValue);
        assert!(delegate.handle.is_observing());

        // Subscribe twice and then unsubscribe the second downstream.
        let downstream2 = new_downstream(RequestId(2), 1);
        let downstream2_id = downstream2.id;
        delegate.subscribe(downstream2).unwrap();
        assert!(!delegate.unsubscribe(downstream2_id, None));
        assert_eq!(txn_extra_op.load(), TxnExtraOp::ReadOldValue);
        assert!(delegate.handle.is_observing());

        // `on_region_ready` when the delegate isn't resolved.
        delegate.subscribe(new_downstream(RequestId(1), 2)).unwrap();
        let mut region = Region::default();
        region.mut_region_epoch().set_conf_ver(1);
        region.mut_region_epoch().set_version(1);
        {
            assert!(delegate.init_lock_tracker());
            let failures = delegate
                .finish_scan_locks(region, Default::default())
                .unwrap();
            assert_eq!(failures.len(), 1);
            let id = failures[0].0.id;
            let result = delegate.unsubscribe_with_scan_cleanup(id, None);
            assert!(result.may_have_queued_scan);
            assert!(!result.delegate_empty);
            assert_eq!(delegate.downstreams().len(), 1);
        }
        assert_eq!(txn_extra_op.load(), TxnExtraOp::ReadOldValue);
        assert!(delegate.handle.is_observing());

        // Subscribe with an invalid epoch.
        delegate
            .subscribe(new_downstream(RequestId(1), 2))
            .unwrap_err();
        assert_eq!(delegate.downstreams().len(), 1);

        // Unsubscribe all downstreams.
        assert!(delegate.unsubscribe(downstream1_id, None));
        assert!(delegate.downstreams().is_empty());
        assert_eq!(txn_extra_op.load(), TxnExtraOp::Noop);
        assert!(!delegate.handle.is_observing());
    }

    #[test]
    fn test_observed_range() {
        for case in [
            (b"".as_slice(), b"".as_slice(), false),
            (b"a", b"", false),
            (b"", b"b", false),
            (b"a", b"b", true),
            (b"a", b"bb", false),
            (b"a", b"aa", true),
            (b"aa", b"aaa", true),
        ] {
            let start_key = if !case.0.is_empty() {
                Key::from_raw(case.0).into_encoded()
            } else {
                case.0.to_owned()
            };
            let end_key = if !case.1.is_empty() {
                Key::from_raw(case.1).into_encoded()
            } else {
                case.1.to_owned()
            };
            let mut region = Region::default();
            region.start_key = start_key.to_owned();
            region.end_key = end_key.to_owned();

            for k in 0..=0xff {
                let mut observed_range = ObservedRange::default();
                observed_range.update_region_key_range(&region);
                assert!(observed_range.contains_encoded_key(&Key::from_raw(&[k]).into_encoded()));
            }
            let mut observed_range = ObservedRange::new(
                Key::from_raw(b"a").into_encoded(),
                Key::from_raw(b"b").into_encoded(),
            )
            .unwrap();
            observed_range.update_region_key_range(&region);
            assert_eq!(observed_range.all_key_covered, case.2, "{:?}", case);
            assert!(
                observed_range.contains_encoded_key(&Key::from_raw(b"a").into_encoded()),
                "{:?}",
                case
            );
            assert!(
                observed_range.contains_encoded_key(&Key::from_raw(b"ab").into_encoded()),
                "{:?}",
                case
            );
            if observed_range.all_key_covered {
                assert!(
                    observed_range.contains_encoded_key(&Key::from_raw(b"b").into_encoded()),
                    "{:?}",
                    case
                );
            } else {
                assert!(
                    !observed_range.contains_encoded_key(&Key::from_raw(b"b").into_encoded()),
                    "{:?}",
                    case
                );
            }
        }
    }

    #[test]
    fn test_downstream_filter_entires() {
        // Create a new delegate that observes [b, d).
        let observed_range = ObservedRange::new(
            Key::from_raw(b"b").into_encoded(),
            Key::from_raw(b"d").into_encoded(),
        )
        .unwrap();
        let memory_quota = Arc::new(MemoryQuota::new(usize::MAX));
        let txn_extra_op = Arc::new(AtomicCell::new(TxnExtraOp::Noop));
        let mut delegate = Delegate::new(1, memory_quota, txn_extra_op);
        assert!(delegate.handle.is_observing());
        assert!(delegate.init_lock_tracker());

        let mut rows_builder = RowsBuilder::default();
        for k in b'a'..=b'e' {
            let mut put = PutRequest::default();
            put.key = Key::from_raw(&[k])
                .append_ts(TimeStamp::new(2))
                .into_encoded();
            put.cf = "write".to_owned();
            put.value = txn_types::Write::new(WriteType::Put, 1.into(), Some(b"test".to_vec()))
                .as_ref()
                .to_bytes();
            delegate.sink_txn_put(put, &mut rows_builder).unwrap();
        }
        assert_eq!(rows_builder.txns_by_key.len(), 5);

        let (sink, mut drain) = channel(ConnId::default(), 1, Arc::new(MemoryQuota::new(1024)));
        let mut downstream = Downstream::new(
            "peer".to_owned(),
            RegionEpoch::default(),
            RequestId(1),
            ConnId::new(),
            false,
            observed_range,
        );
        downstream.set_sink(sink);
        downstream.get_state().store(DownstreamState::Normal);
        delegate.add_downstream(downstream);
        let entries = rows_builder.finish_build();
        delegate
            .sink_downstream_tidb(entries, |_, _| Ok(()))
            .unwrap();

        let (mut tx, mut rx) = futures::channel::mpsc::unbounded();
        let runtime = tokio::runtime::Runtime::new().unwrap();
        runtime.spawn(async move {
            drain.forward(&mut tx, None).await.unwrap();
        });
        let (e, _) = recv_timeout(&mut rx, std::time::Duration::from_secs(5))
            .unwrap()
            .unwrap();
        assert_eq!(e.events[0].get_entries().get_entries().len(), 2, "{:?}", e);
    }

    fn test_downstream_txn_source_filter(txn_source: TxnSource, filter_loop: bool) {
        // Create a new delegate that observes [a, f).
        let observed_range = ObservedRange::new(
            Key::from_raw(b"a").into_encoded(),
            Key::from_raw(b"f").into_encoded(),
        )
        .unwrap();
        let memory_quota = Arc::new(MemoryQuota::new(usize::MAX));
        let txn_extra_op = Arc::new(AtomicCell::new(TxnExtraOp::Noop));
        let mut delegate = Delegate::new(1, memory_quota, txn_extra_op);
        assert!(delegate.handle.is_observing());
        assert!(delegate.init_lock_tracker());

        let mut rows_builder = RowsBuilder::default();
        for k in b'a'..=b'e' {
            let mut put = PutRequest::default();
            put.key = Key::from_raw(&[k])
                .append_ts(TimeStamp::new(2))
                .into_encoded();
            put.cf = "write".to_owned();
            let mut write = txn_types::Write::new(WriteType::Put, 1.into(), Some(b"test".to_vec()));
            // Only the key `a` is a normal write.
            if k != b'a' {
                write = write.set_txn_source(txn_source.into());
            }
            put.value = write.as_ref().to_bytes();
            delegate.sink_txn_put(put, &mut rows_builder).unwrap();
        }
        assert_eq!(rows_builder.txns_by_key.len(), 5);

        let (sink, mut drain) = channel(ConnId::default(), 1, Arc::new(MemoryQuota::new(1024)));
        let mut downstream = Downstream::new(
            "peer".to_owned(),
            RegionEpoch::default(),
            RequestId(1),
            ConnId::new(),
            filter_loop,
            observed_range,
        );
        downstream.set_sink(sink);
        downstream.get_state().store(DownstreamState::Normal);
        delegate.add_downstream(downstream);
        let entries = rows_builder.finish_build();
        delegate
            .sink_downstream_tidb(entries, |_, _| Ok(()))
            .unwrap();

        let (mut tx, mut rx) = futures::channel::mpsc::unbounded();
        let runtime = tokio::runtime::Runtime::new().unwrap();
        runtime.spawn(async move {
            drain.forward(&mut tx, None).await.unwrap();
        });
        let (e, _) = recv_timeout(&mut rx, std::time::Duration::from_secs(5))
            .unwrap()
            .unwrap();
        assert_eq!(e.events[0].get_entries().get_entries().len(), 1, "{:?}", e);
    }

    #[test]
    fn test_downstream_filter_cdc_write_entires() {
        let mut txn_source = TxnSource::default();
        txn_source.set_cdc_write_source(1);

        test_downstream_txn_source_filter(txn_source, true);
    }

    #[test]
    fn test_downstream_filter_lightning_physical_import() {
        let mut txn_source = TxnSource::default();
        txn_source.set_lightning_physical_import();
        test_downstream_txn_source_filter(txn_source, false);
    }

    #[test]
    fn test_downstream_filter_lossy_ddl_entires() {
        let mut txn_source = TxnSource::default();
        txn_source.set_lossy_ddl_reorg_source(1);
        test_downstream_txn_source_filter(txn_source, false);

        // With cdr write source and filter loop is false, we should still ignore lossy
        // ddl changes.
        let mut txn_source = TxnSource::default();
        txn_source.set_cdc_write_source(1);
        txn_source.set_lossy_ddl_reorg_source(1);
        test_downstream_txn_source_filter(txn_source, false);

        // With cdr write source and filter loop is true, we should still ignore some
        // events.
        let mut txn_source = TxnSource::default();
        txn_source.set_cdc_write_source(1);
        txn_source.set_lossy_ddl_reorg_source(1);
        test_downstream_txn_source_filter(txn_source, true);
    }

    #[test]
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    fn test_decode_rawkv() {
        let cases = vec![
            (vec![b'r', 2, 3, 4], b"world1".to_vec(), 1, Some(10), false),
            (vec![b'r', 3, 4, 5], b"world2".to_vec(), 2, None, true),
        ];

        for (key, value, start_ts, expire_ts, is_delete) in cases.into_iter() {
            let mut row = EventRow::default();
            let key_with_ts = ApiV2::encode_raw_key(&key, Some(start_ts.into())).into_encoded();
            let raw_value = RawValue {
                user_value: value.to_vec(),
                expire_ts,
                is_delete,
            };
            let encoded_value = ApiV2::encode_raw_value_owned(raw_value);
            decode_rawkv(key_with_ts, encoded_value, &mut row).unwrap();

            assert_eq!(row.start_ts, start_ts);
            assert_eq!(row.commit_ts, start_ts);
            assert_eq!(row.key, key);
            assert_eq!(row.value, value);

            if is_delete {
                assert_eq!(row.op_type, EventRowOpType::Delete);
            } else {
                assert_eq!(row.op_type, EventRowOpType::Put);
            }

            if let Some(expire_ts) = expire_ts {
                assert_eq!(row.expire_ts_unix_secs, expire_ts);
            } else {
                assert_eq!(row.expire_ts_unix_secs, 0);
            }
        }
    }

    #[test]
    fn test_lock_tracker() {
        let quota = Arc::new(MemoryQuota::new(usize::MAX));
        let mut delegate = Delegate::new(1, quota.clone(), Default::default());
        assert!(delegate.init_lock_tracker());
        assert!(!delegate.init_lock_tracker());

        let mut k1 = Vec::with_capacity(100);
        k1.extend_from_slice(Key::from_raw(b"key1").as_encoded());
        let k1 = Key::from_encoded(k1);
        assert_eq!(delegate.push_lock(k1, MiniLock::from_ts(100)).unwrap(), 0);
        assert_eq!(quota.in_use(), 100);

        delegate
            .pop_lock(Key::from_raw(b"key1"), TimeStamp::from(99))
            .unwrap();
        assert_eq!(quota.in_use(), 117);

        delegate
            .pop_lock(Key::from_raw(b"key1"), TimeStamp::from(100))
            .unwrap();
        assert_eq!(quota.in_use(), 134);

        let mut k2 = Vec::with_capacity(200);
        k2.extend_from_slice(Key::from_raw(b"key2").as_encoded());
        let k2 = Key::from_encoded(k2);
        assert_eq!(delegate.push_lock(k2, MiniLock::from_ts(100)).unwrap(), 0);
        assert_eq!(quota.in_use(), 334);

        let mut scaned_locks = BTreeMap::default();
        scaned_locks.insert(Key::from_raw(b"key1"), MiniLock::from_ts(100));
        scaned_locks.insert(Key::from_raw(b"key2"), MiniLock::from_ts(100));
        scaned_locks.insert(Key::from_raw(b"key3"), MiniLock::from_ts(100));
        delegate
            .finish_prepare_lock_tracker(Default::default(), scaned_locks)
            .unwrap();
        assert_eq!(quota.in_use(), 34);

        delegate
            .pop_lock(Key::from_raw(b"key2"), TimeStamp::from(100))
            .unwrap();
        delegate
            .pop_lock(Key::from_raw(b"key3"), TimeStamp::from(100))
            .unwrap();
        assert_eq!(quota.in_use(), 0);

        let v = delegate
            .push_lock(Key::from_raw(b"key1"), MiniLock::from_ts(300))
            .unwrap();
        assert_eq!(v, 1);
        assert_eq!(quota.in_use(), 17);
        let v = delegate
            .push_lock(Key::from_raw(b"key1"), MiniLock::from_ts(300))
            .unwrap();
        assert_eq!(v, 0);
        assert_eq!(quota.in_use(), 17);
    }

    #[test]
    fn test_lock_tracker_untrack_vacant() {
        let quota = Arc::new(MemoryQuota::new(usize::MAX));
        let mut delegate = Delegate::new(1, quota.clone(), Default::default());
        assert!(delegate.init_lock_tracker());
        assert!(!delegate.init_lock_tracker());

        delegate
            .pop_lock(Key::from_raw(b"key1"), TimeStamp::zero())
            .unwrap();
        let mut scaned_locks = BTreeMap::default();
        scaned_locks.insert(Key::from_raw(b"key2"), MiniLock::from_ts(100));
        delegate
            .finish_prepare_lock_tracker(Default::default(), scaned_locks)
            .unwrap();
    }

    #[test]
    fn test_lock_tracker_preparing_duplicate_track_is_idempotent() {
        let quota = Arc::new(MemoryQuota::new(usize::MAX));
        let mut delegate = Delegate::new(1, quota.clone(), Default::default());
        assert!(delegate.init_lock_tracker());

        let key = Key::from_raw(b"txn_file_key");
        assert_eq!(
            delegate
                .push_lock(key.clone(), MiniLock::new(10, 1))
                .unwrap(),
            0
        );
        let after_first_track = quota.in_use();
        assert_eq!(
            delegate
                .push_lock(key.clone(), MiniLock::new(10, 2))
                .unwrap(),
            0
        );
        assert_eq!(quota.in_use(), after_first_track * 2);

        delegate
            .finish_prepare_lock_tracker(Default::default(), BTreeMap::default())
            .unwrap();
        assert_eq!(delegate.push_lock(key, MiniLock::new(10, 3)).unwrap(), 0);
        assert_eq!(quota.in_use(), after_first_track);
    }
}