use std::{
    collections::{HashMap, VecDeque},
    sync::Arc,
    time::Instant,
};

use anyhow::{Context, Result, bail, ensure};
use log::{error, info};
use storage_backend::storage::BackendStorage;
use tici_common::cdc_file_name::CDCFileName;
use tokio::{task::JoinHandle, time::Duration};
use tokio_util::sync::CancellationToken;

use crate::{
    CDCCheckpoint,
    utils::{
        metrics::{inc_cdc_event_count, inc_cdc_op_count_by, start_cdc_op_timer},
        tantivy_doc::TantivyDoc,
    },
    writer::{
        cdc_file_cache::{CDCFileCache, FileContent},
        cdc_file_prefix::CDCPrefix,
        cdc_file_reader_state::{ReaderState, ShardInfo},
        cdc_list_dispatcher::CDCListDispatcher,
        cdc_log::{CDCLog, CDCLogEntry, RowDataAccess},
    },
};

// CDCFileReader reads CDC files and dispatches
// the logs to corresponding shard writers based on shard ranges.
pub struct CDCFileReader {
    bucket: String,
    file_cache: Arc<CDCFileCache>,
    prefetch_window: usize,
    cdc_poll_interval: Duration,
    cdc_max_poll_interval: Duration,
    list_dispatcher: Arc<CDCListDispatcher>,
    state: Arc<ReaderState>,
    background_handle: Option<JoinHandle<Result<()>>>,
    cancel_token: CancellationToken,
    prefix: CDCPrefix,
}

type PrefetchTask = (String, JoinHandle<Result<FileContent>>);

const EMPTY_POLL_BACKOFF_MULTIPLIER: u32 = 2;

#[derive(Debug, Clone)]
struct EmptyPollBackoff {
    base_delay: Duration,
    max_delay: Duration,
    current_delay: Duration,
}

impl EmptyPollBackoff {
    fn new(base_delay: Duration, max_delay: Duration) -> Self {
        Self {
            base_delay,
            max_delay,
            current_delay: base_delay,
        }
    }

    fn current_delay(&self) -> Duration {
        self.current_delay
    }

    fn on_empty_poll(&mut self) -> Duration {
        self.current_delay = self
            .current_delay
            .saturating_mul(EMPTY_POLL_BACKOFF_MULTIPLIER)
            .min(self.max_delay);
        self.current_delay
    }

    fn reset(&mut self) {
        self.current_delay = self.base_delay;
    }
}

fn cdc_line_error_context(line: &str, line_num: usize, filename: &CDCFileName) -> String {
    format!(
        "line={}, line_num={}, filename={}",
        line, line_num, filename
    )
}

impl CDCFileReader {
    fn log_background_stop_result(
        join_result: std::result::Result<Result<()>, tokio::task::JoinError>,
    ) {
        match join_result {
            Ok(Ok(())) => info!("Background task stopped successfully"),
            Ok(Err(e)) => error!("Background task stop error: {:?}", e),
            Err(e) => {
                if e.is_cancelled() {
                    info!("Background task cancelled successfully");
                } else {
                    error!("Background task stop error: {:?}", e);
                }
            }
        }
    }

    #[cfg(test)]
    pub fn new(
        backend_storage: Arc<dyn BackendStorage>,
        cdc_checkpoint: &CDCCheckpoint,
        cdc_poll_interval: Duration,
        cdc_max_poll_interval: Duration,
        prefetch_window: usize,
        head_object_last_modified_fallback: bool,
    ) -> Self {
        let list_dispatcher = Arc::new(CDCListDispatcher::new(backend_storage.clone(), 1));
        Self::new_with_dispatcher(
            backend_storage,
            cdc_checkpoint,
            cdc_poll_interval,
            cdc_max_poll_interval,
            prefetch_window,
            head_object_last_modified_fallback,
            list_dispatcher,
        )
    }

    pub fn new_with_dispatcher(
        backend_storage: Arc<dyn BackendStorage>,
        cdc_checkpoint: &CDCCheckpoint,
        cdc_poll_interval: Duration,
        cdc_max_poll_interval: Duration,
        prefetch_window: usize,
        head_object_last_modified_fallback: bool,
        list_dispatcher: Arc<CDCListDispatcher>,
    ) -> Self {
        assert!(
            prefetch_window > 0,
            "cdc_prefetch_window must be greater than 0"
        );
        let file_cache = Arc::new(CDCFileCache::new(
            backend_storage.clone(),
            cdc_checkpoint.bucket.clone(),
            head_object_last_modified_fallback,
        ));
        let state = ReaderState::new(cdc_checkpoint.last_file.clone());
        CDCFileReader {
            bucket: cdc_checkpoint.bucket.clone(),
            file_cache,
            prefetch_window,
            cdc_poll_interval,
            cdc_max_poll_interval,
            list_dispatcher,
            state: Arc::new(state),
            background_handle: None,
            cancel_token: CancellationToken::new(),
            prefix: CDCPrefix::new(cdc_checkpoint.table_prefix.clone()),
        }
    }

    pub fn get_prefix(&self) -> &CDCPrefix {
        &self.prefix
    }

    pub fn get_state(&self) -> &Arc<ReaderState> {
        &self.state
    }

    pub fn add_shard(&self, shard_info: ShardInfo) {
        let shard_id = shard_info.shard_id;
        self.state.add_shard_info(shard_id, Arc::new(shard_info));
    }

    pub fn remove_shard(&self, shard_id: u64) {
        if let Some(shard_info) = self.state.remove_shard_info(shard_id) {
            info!("Removed shard {}", shard_info.tracing_id);
        }
    }

    pub fn is_empty(&self) -> bool {
        self.state.is_shard_infos_empty()
    }

    fn spawn_prefetch_task(
        file_cache: Arc<CDCFileCache>,
        filename: &str,
    ) -> JoinHandle<Result<FileContent>> {
        let filename = filename.to_string();
        tokio::spawn(async move { file_cache.prefetch_file(&filename).await })
    }

    fn fill_prefetch_tasks(
        file_cache: &Arc<CDCFileCache>,
        unread_files: &[String],
        next_prefetch_idx: &mut usize,
        prefetch_window: usize,
        inflight_prefetch: &mut VecDeque<PrefetchTask>,
    ) {
        // Each inflight task owns at most one prefetched file once the S3 read
        // completes. Keeping at most `prefetch_window` tasks here therefore
        // also bounds prefetched file memory; CDCFileCache intentionally does
        // not keep a second in-memory copy.
        let available_slots = prefetch_window.saturating_sub(inflight_prefetch.len());
        let end_idx = (*next_prefetch_idx + available_slots).min(unread_files.len());
        for filename in unread_files[*next_prefetch_idx..end_idx].iter().cloned() {
            let prefetch_task = Self::spawn_prefetch_task(file_cache.clone(), &filename);
            inflight_prefetch.push_back((filename, prefetch_task));
        }
        *next_prefetch_idx = end_idx;
    }

    fn abort_prefetch_tasks(inflight_prefetch: &mut VecDeque<PrefetchTask>) {
        inc_cdc_op_count_by("abort_prefetch_task", inflight_prefetch.len() as u64);
        for (_, prefetch_task) in inflight_prefetch.drain(..) {
            prefetch_task.abort();
        }
    }

    async fn process_unread_files(
        file_cache: &Arc<CDCFileCache>,
        unread_files: Vec<String>,
        poll_prev_file: CDCFileName,
        state: &Arc<ReaderState>,
        cancel_token: &CancellationToken,
        prefetch_window: usize,
    ) -> Result<()> {
        // Keep using the same prev_file snapshot that was passed to
        // list_unread_files() in this poll round. Re-reading shared state here
        // can race with add_shard() rewinding the cursor and cause us to skip
        // intermediate CDC files for newly added lagging shards.
        let mut prev_file = poll_prev_file;
        let mut next_prefetch_idx = 0usize;
        let mut inflight_prefetch = VecDeque::new();

        Self::fill_prefetch_tasks(
            file_cache,
            &unread_files,
            &mut next_prefetch_idx,
            prefetch_window,
            &mut inflight_prefetch,
        );

        let result = async {
            while let Some((filename, prefetch_task)) = inflight_prefetch.pop_front() {
                if cancel_token.is_cancelled() {
                    prefetch_task.abort();
                    return Ok(());
                }
                let prefetch_wait_start = Instant::now();
                let mut file_content = prefetch_task
                    .await
                    .with_context(|| format!("Prefetch task join failed for {}", filename))??;
                file_content.stats.prefetch_wait_elapsed =
                    Some(prefetch_wait_start.elapsed().as_secs_f64());
                if cancel_token.is_cancelled() {
                    return Ok(());
                }

                let filename = CDCFileName::parse(&filename)?;
                Self::process_file(file_content, &filename, &prev_file, state).await?;

                let can_advance = state.check_and_update_last_file(&filename)?;
                if !can_advance {
                    break;
                }
                prev_file = filename;

                Self::fill_prefetch_tasks(
                    file_cache,
                    &unread_files,
                    &mut next_prefetch_idx,
                    prefetch_window,
                    &mut inflight_prefetch,
                );
            }
            Ok(())
        }
        .await;

        Self::abort_prefetch_tasks(&mut inflight_prefetch);
        result
    }

    async fn process_file(
        file_content: FileContent,
        filename: &CDCFileName,
        prev_file: &CDCFileName,
        state: &Arc<ReaderState>,
    ) -> Result<()> {
        let timer = start_cdc_op_timer("process_file");
        let FileContent { data, stats } = file_content;
        let mut total_cdc_events = 0;
        let mut not_dml_cdc_events = 0;
        let mut shards_for_processing = state.get_shards_for_processing(prev_file);
        if shards_for_processing.is_empty() {
            info!(
                "Skip processing {}, no shards ready for prev_file={}",
                filename, prev_file
            );
            return Ok(());
        }
        // CDC file entries are ordered by version, so we can activate index groups
        // incrementally once the stream reaches their created_ts threshold.
        shards_for_processing.sort_unstable_by_key(|shard_info| shard_info.index_meta.created_ts());

        // Count total shards for early exit check.
        let total_shard_count: usize = shards_for_processing
            .iter()
            .map(|idx| idx.get_shard_count())
            .sum();
        let mut closed_shard_count = 0usize;
        let mut active_index_shard_end = 0usize;

        let mut shard_dispatch_count: HashMap<u64, u64> = HashMap::new();
        let mut index_filtered_count = vec![0; shards_for_processing.len()];
        let mut total_blocked_secs = 0.0;
        // Process each line as a separate JSON object
        for (line_num, line) in data.lines().enumerate() {
            if line.trim().is_empty() {
                continue;
            }
            inc_cdc_event_count("total");
            total_cdc_events += 1;
            let cdc_log = match CDCLog::new(line) {
                Ok(Some(cdc_log)) => cdc_log,
                Ok(None) => {
                    not_dml_cdc_events += 1;
                    inc_cdc_event_count("not_dml");
                    info!(
                        "Skip {} in file={}, line_num={}, not_dml_cdc_events_count={}",
                        line, filename, line_num, not_dml_cdc_events
                    );
                    continue;
                }
                Err(e) => {
                    inc_cdc_event_count("parse_err");
                    bail!(
                        "Failed to parse {} in file={}, line_num={}, err={:?}",
                        line,
                        line_num,
                        filename,
                        e
                    );
                }
            };
            let version = cdc_log
                .get_version()
                .inspect_err(|_| inc_cdc_event_count("version_not_found"))
                .with_context(|| cdc_line_error_context(line, line_num, filename))?;

            // Try activate index shards incrementally once the stream reaches their
            // created_ts threshold.
            while active_index_shard_end < shards_for_processing.len()
                && shards_for_processing[active_index_shard_end]
                    .index_meta
                    .created_ts()
                    <= version
            {
                active_index_shard_end += 1;
            }
            for count in &mut index_filtered_count[active_index_shard_end..] {
                inc_cdc_event_count("stale_cdc_log");
                *count += 1;
            }
            if active_index_shard_end == 0 {
                continue;
            }

            for index_shard_info in shards_for_processing[..active_index_shard_end].iter() {
                let sharding_key_pair = cdc_log
                    .get_sharding_key_pair(&index_shard_info.index_meta)
                    .inspect_err(|_| inc_cdc_event_count("get_sharding_key_err"))
                    .with_context(|| cdc_line_error_context(line, line_num, filename))?;
                let data_sharding_key = sharding_key_pair.data_sharding_key;
                let old_sharding_key = sharding_key_pair.old_sharding_key;

                let data_shard_infos = index_shard_info
                    .get_shard_infos_by_key(data_sharding_key.raw())
                    .into_iter()
                    .filter(|shard_info| !shard_info.is_closed())
                    .collect::<Vec<_>>();
                let data_tantivy_docs = data_shard_infos
                    .iter()
                    .map(|shard_info| {
                        TantivyDoc::new(&cdc_log, &shard_info.index_meta)
                            .inspect_err(|_| inc_cdc_event_count("new_tantivy_doc_err"))
                            .with_context(|| cdc_line_error_context(line, line_num, filename))
                    })
                    .collect::<Result<Vec<TantivyDoc>>>()?;

                let old_shard_infos = if let Some(old_sharding_key) = old_sharding_key.as_ref() {
                    index_shard_info
                        .get_shard_infos_by_key(old_sharding_key.raw())
                        .into_iter()
                        .filter(|shard_info| !shard_info.is_closed())
                        .collect::<Vec<_>>()
                } else {
                    Vec::new()
                };
                let old_tantivy_docs = old_shard_infos
                    .iter()
                    .map(|_shard_info| {
                        if let Some(old_sharding_key) = old_sharding_key.as_ref() {
                            Ok(TantivyDoc::delete_doc(old_sharding_key.clone(), version))
                        } else {
                            bail!(
                                "old_sharding_key is None, line={}, line_num={}, filename={}",
                                line,
                                line_num,
                                filename
                            );
                        }
                    })
                    .collect::<Result<Vec<TantivyDoc>>>()?;

                ensure!(
                    old_tantivy_docs.len() == old_shard_infos.len(),
                    "old_tantivy_docs={}, old_shard_infos={}",
                    old_tantivy_docs.len(),
                    old_shard_infos.len()
                );
                for (shard_info, delete_doc) in old_shard_infos
                    .into_iter()
                    .zip(old_tantivy_docs.into_iter())
                {
                    let (success, blocked_secs) =
                        shard_info.send(CDCLogEntry::CDCData(delete_doc)).await;
                    total_blocked_secs += blocked_secs;
                    if !success {
                        closed_shard_count += 1;
                        if closed_shard_count == total_shard_count {
                            info!(
                                "All shards closed, abort processing {} at line {}",
                                filename, line_num
                            );
                            return Ok(());
                        }
                    }
                    *shard_dispatch_count.entry(shard_info.shard_id).or_default() += 1;
                }

                ensure!(
                    data_tantivy_docs.len() == data_shard_infos.len(),
                    "data_tantivy_docs={}, data_shard_infos={}",
                    data_tantivy_docs.len(),
                    data_shard_infos.len()
                );
                for (shard_info, data_doc) in data_shard_infos
                    .into_iter()
                    .zip(data_tantivy_docs.into_iter())
                {
                    let (success, blocked_secs) =
                        shard_info.send(CDCLogEntry::CDCData(data_doc)).await;
                    total_blocked_secs += blocked_secs;
                    if !success {
                        closed_shard_count += 1;
                        if closed_shard_count == total_shard_count {
                            info!(
                                "All shards closed, abort processing {} at line {}",
                                filename, line_num
                            );
                            return Ok(());
                        }
                    }
                    *shard_dispatch_count.entry(shard_info.shard_id).or_default() += 1;
                }
            }
        }

        let filename_entry = CDCLogEntry::Filename(filename.clone());
        for index_shard_info in shards_for_processing.iter() {
            let shard_infos = index_shard_info.get_all_shard_infos();
            for shard_info in shard_infos {
                if shard_info.is_closed() {
                    continue;
                }
                let (success, blocked_secs) = shard_info.send(filename_entry.clone()).await;
                total_blocked_secs += blocked_secs;
                if !success {
                    continue;
                }
                shard_info.update_last_file(filename.clone());
            }
        }

        info!(
            concat!(
                "Processed {}, total={}, not_dml={}, total_shards={}, closed_shards={}, ",
                "shard_dispatch={:?}, index_filtered={:?}, bytes={}, get_object={:.3}s, ",
                "read_object={:.3}s, utf8_decode={:.3}s, downstream_delay={:?}s, ",
                "prefetch_wait={:?}s, blocked={:.3}s, cost={:.3}s"
            ),
            filename,
            total_cdc_events,
            not_dml_cdc_events,
            total_shard_count,
            closed_shard_count,
            shard_dispatch_count,
            index_filtered_count,
            stats.bytes_len,
            stats.get_object_elapsed,
            stats.read_object_elapsed,
            stats.utf8_decode_elapsed,
            stats.downstream_delay.unwrap_or_default(),
            stats.prefetch_wait_elapsed.unwrap_or_default(),
            total_blocked_secs,
            timer.stop_and_record(),
        );
        Ok(())
    }

    pub fn start(&mut self) {
        if self.background_handle.is_some() {
            info!("Already started: {:?}", self.prefix);
            return;
        }

        let file_cache = self.file_cache.clone();

        // In path: "default_prefix/cdc/123/463728960809467906/CDC*.json",
        // 'prefix' value is: "default_prefix/cdc/123".
        let bucket = self.bucket.clone();
        let prefix = self.get_prefix().clone();
        let poll_interval = self.cdc_poll_interval;
        let max_poll_interval = self.cdc_max_poll_interval;
        let prefetch_window = self.prefetch_window;
        let list_dispatcher = self.list_dispatcher.clone();
        let state = self.state.clone();
        let cancel_token = self.cancel_token.clone();
        let handle = tokio::spawn(async move {
            info!("Background task started: {:?}", prefix);

            let mut empty_poll_backoff = EmptyPollBackoff::new(poll_interval, max_poll_interval);
            let mut should_sleep = false;
            loop {
                if should_sleep {
                    tokio::select! {
                        _ = cancel_token.cancelled() => {
                            info!("Stop signal received, exiting background task: {:?}", prefix);
                            return Ok(());
                        }

                        _ = tokio::time::sleep(empty_poll_backoff.current_delay()) => {}
                    }
                }

                let task = async {
                    let poll_prev_file = state.get_last_file();
                    let unread_files = list_dispatcher
                        .list_unread_files(&bucket, &poll_prev_file, &prefix)
                        .await?;
                    if unread_files.is_empty() {
                        empty_poll_backoff.on_empty_poll();
                        return Ok::<bool, anyhow::Error>(false);
                    }
                    empty_poll_backoff.reset();

                    Self::process_unread_files(
                        &file_cache,
                        unread_files,
                        poll_prev_file,
                        &state,
                        &cancel_token,
                        prefetch_window,
                    )
                    .await?;

                    Ok::<bool, anyhow::Error>(true)
                };
                should_sleep = match task.await {
                    Ok(has_unread_files) => !has_unread_files,
                    Err(e) => {
                        error!(
                            "CDC file poll/process error on prefix={:?}, error={:?}",
                            prefix, e
                        );
                        true
                    }
                };
            }
        });

        self.background_handle = Some(handle);
    }

    pub async fn stop(mut self) {
        if let Some(handle) = self.background_handle.take() {
            handle.abort();
            Self::log_background_stop_result(handle.await);
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{sync::Arc, time::Duration};

    use serde_json::json;
    use storage_backend::storage::create_backend_storage_for_test;
    use tici_common::{cdc_file_name::CDC_FILENAME_MATCH, config::S3Config, index::IndexKey};
    use tici_grpc::ShardManifestHeader;
    use tici_test_utils::test_utils::init_logger_for_ut;
    use tokio::{sync::mpsc, time::timeout};

    use super::*;
    use crate::{
        CDCCheckpoint, FragLocation, ShardWriterMeta, ShardWriterMetaVersion,
        test_utils::{
            MockBackendStorage, TEST_HDFSLOGS_INDEX_META, TEST_INDEX_KEY, TEST_INF_END,
            TEST_INF_START, TEST_S3_BUCKET, TEST_S3_BUCKET_PARTITION, TEST_S3_PREFIX,
            TEST_TABLE_PREFIX, mock_custom_employees_partition_index_meta,
            mock_custom_hdfs_logs_index_meta, test_create_backend_storage,
        },
        utils::{key::ShardingKeyRange, metrics::get_cdc_event_count, schema::IndexMeta},
        writer::{cdc_file_poller::CDCFilePoller, cdc_list_dispatcher::CDCListDispatcher},
    };

    const TEST_CDC_PREFETCH_WINDOW: usize = 2;
    const TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK: bool = true;

    impl CDCFileReader {
        fn get_shard_info(&self, shard_id: u64) -> Option<Arc<ShardInfo>> {
            self.state.get_shard_info(shard_id)
        }

        fn contains_shard(&self, shard_id: u64) -> bool {
            self.state.contains_shard(shard_id)
        }

        fn get_shard_last_file(&self, shard_id: u64) -> Option<CDCFileName> {
            self.get_shard_info(shard_id)
                .map(|shard_info| shard_info.get_last_file())
        }
    }

    fn mock_hdfs_logs_index_meta_with_created_ts(
        index_key: IndexKey,
        created_ts: u64,
    ) -> IndexMeta {
        let index_meta = mock_custom_hdfs_logs_index_meta(index_key, &[]).unwrap();
        let mut index_meta_value = serde_json::to_value(index_meta).unwrap();
        index_meta_value["created_ts"] = json!(created_ts);
        serde_json::from_value(index_meta_value).unwrap()
    }

    fn mock_insert_cdc_log_line(version: u64, row_id: u64, body: &str) -> String {
        format!(
            r#"{{"id":0,"database":"test","table":"hdfs_15w","isDdl":false,"type":"INSERT","data":[{{"id":"{row_id}","timestamp":"1460530013","severity_text":"INFO","body":"{body}","tenant_id":"57"}}],"_tidb":{{"commitTs":{version},"rowkey":"dIAAAAAAAAB7X3KAAAAAAAAD6Q=="}}}}"#
        )
    }

    fn drain_receiver(rx: &mut mpsc::Receiver<CDCLogEntry>) -> (usize, Vec<String>) {
        let mut cdc_data_count = 0;
        let mut filenames = Vec::new();
        while let Ok(entry) = rx.try_recv() {
            match entry {
                CDCLogEntry::CDCData(_) => cdc_data_count += 1,
                CDCLogEntry::Filename(fname) => filenames.push(fname.to_string()),
            }
        }
        (cdc_data_count, filenames)
    }

    fn mock_shard_meta(shard_id: u64, last_file: &str) -> ShardWriterMeta {
        ShardWriterMeta {
            index_key: IndexKey::new_nullspace(0, 0),
            version: ShardWriterMetaVersion::LATEST,
            shard_header: ShardManifestHeader {
                start_key: TEST_INF_START.to_vec(),
                end_key: TEST_INF_END.to_vec(),
                shard_id,
                epoch: 0,
                seq: 0,
            },
            cdc_checkpoint: CDCCheckpoint {
                bucket: TEST_S3_BUCKET.to_string(),
                table_prefix: "cdc_table_prefix".to_string(),
                last_file: CDCFileName::parse(last_file).unwrap(),
            },
            frag_location: FragLocation {
                bucket: TEST_S3_BUCKET.to_string(),
                prefix: TEST_S3_PREFIX.to_string(),
            },
        }
    }

    fn mock_shard_meta_with_checkpoint(
        shard_id: u64,
        cdc_checkpoint: CDCCheckpoint,
    ) -> ShardWriterMeta {
        ShardWriterMeta {
            index_key: IndexKey::new_nullspace(0, 0),
            version: ShardWriterMetaVersion::LATEST,
            shard_header: ShardManifestHeader {
                start_key: TEST_INF_START.to_vec(),
                end_key: TEST_INF_END.to_vec(),
                shard_id,
                epoch: 0,
                seq: 0,
            },
            cdc_checkpoint,
            frag_location: FragLocation {
                bucket: TEST_S3_BUCKET.to_string(),
                prefix: TEST_S3_PREFIX.to_string(),
            },
        }
    }

    fn mock_reader_shard_info(
        shard_meta: &ShardWriterMeta,
        sender: mpsc::Sender<CDCLogEntry>,
        index_meta: IndexMeta,
    ) -> ShardInfo {
        let shard_range = ShardingKeyRange::half_open(
            shard_meta.shard_header.start_key.clone(),
            shard_meta.shard_header.end_key.clone(),
        )
        .unwrap();
        ShardInfo::new(
            shard_meta.shard_header.shard_id,
            sender,
            shard_range,
            shard_meta.shard_tracing_id(),
            shard_meta.cdc_checkpoint.last_file.clone(),
            index_meta,
        )
        .unwrap()
    }

    fn mock_cdc_file_name(is_partition_table: bool) -> (String, String) {
        if is_partition_table {
            (
                "tici_default_prefix/cdc/428/461869219056451591/CDC00000000000000000001.json"
                    .to_string(),
                "tici_default_prefix/cdc/428/461869219056451591/CDC00000000000000000002.json"
                    .to_string(),
            )
        } else {
            (
                "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json"
                    .to_string(),
                "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000002.json"
                    .to_string(),
            )
        }
    }

    async fn _test_add_and_remove_shard((cdc_file_name1, cdc_file_name2): (String, String)) {
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let config = S3Config::default();
        let backend_storage = create_backend_storage_for_test(&config).await.unwrap();
        let cdc_checkpoint = CDCCheckpoint {
            bucket: "bucket".to_string(),
            table_prefix: "prefix".to_string(),
            last_file: CDCFileName::parse(&cdc_file_name1).unwrap(),
        };
        let reader = CDCFileReader::new(
            backend_storage,
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );
        let (tx, _) = mpsc::channel(1);
        let meta = mock_shard_meta(1, &cdc_file_name1);
        reader.add_shard(mock_reader_shard_info(
            &meta,
            tx.clone(),
            index_meta.clone(),
        ));
        {
            let shard_info = reader.get_shard_info(1).unwrap();
            let last_file = shard_info.get_last_file();
            assert_eq!(last_file, CDCFileName::parse(&cdc_file_name1).unwrap());
            // modify last_file to 2
            shard_info.update_last_file(CDCFileName::parse(&cdc_file_name2).unwrap());
        }
        reader.add_shard(mock_reader_shard_info(&meta, tx.clone(), index_meta));
        {
            let shard_info = reader.get_shard_info(1).unwrap();
            let last_file = shard_info.get_last_file();
            // should update last_file back to 1
            assert_eq!(last_file, CDCFileName::parse(&cdc_file_name1).unwrap());
        }
        reader.remove_shard(1);
        assert!(!reader.contains_shard(1));
    }

    #[tokio::test]
    async fn test_add_and_remove_shard() {
        _test_add_and_remove_shard(mock_cdc_file_name(false)).await;
        _test_add_and_remove_shard(mock_cdc_file_name(true)).await;
    }

    async fn _test_cdc_file_reader_is_empty((cdc_file_name1, _cdc_file_name2): (String, String)) {
        let config = S3Config::default();
        let backend_storage = create_backend_storage_for_test(&config).await.unwrap();
        let cdc_checkpoint = CDCCheckpoint {
            bucket: "bucket".to_string(),
            table_prefix: "prefix".to_string(),
            last_file: CDCFileName::parse(&cdc_file_name1).unwrap(),
        };
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let reader = CDCFileReader::new(
            backend_storage,
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        assert!(reader.is_empty());

        let (tx, _) = mpsc::channel(1);
        let meta = mock_shard_meta(1, &cdc_file_name1);
        reader.add_shard(mock_reader_shard_info(&meta, tx, index_meta));

        assert!(!reader.is_empty());

        reader.remove_shard(1);
        assert!(reader.is_empty());
    }

    #[tokio::test]
    async fn test_cdc_file_reader_is_empty() {
        _test_cdc_file_reader_is_empty(mock_cdc_file_name(false)).await;
        _test_cdc_file_reader_is_empty(mock_cdc_file_name(true)).await;
    }

    async fn _test_multiple_shards_add_remove((cdc_file_name1, _cdc_file_name2): (String, String)) {
        let config = S3Config::default();
        let backend_storage = create_backend_storage_for_test(&config).await.unwrap();
        let cdc_checkpoint = CDCCheckpoint {
            bucket: "bucket".to_string(),
            table_prefix: "prefix".to_string(),
            last_file: CDCFileName::parse(&cdc_file_name1).unwrap(),
        };
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let reader = CDCFileReader::new(
            backend_storage,
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        // Add multiple shards
        for shard_id in 1..=5 {
            let (tx, _) = mpsc::channel(1);
            let meta = mock_shard_meta(shard_id, &cdc_file_name1);
            reader.add_shard(mock_reader_shard_info(&meta, tx, index_meta.clone()));
            assert!(reader.contains_shard(shard_id));
        }

        // Verify all shards exist
        for shard_id in 1..=5 {
            assert!(reader.contains_shard(shard_id));
        }

        // Remove shards one by one
        for shard_id in 1..=5 {
            reader.remove_shard(shard_id);
            assert!(!reader.contains_shard(shard_id));
        }

        assert!(reader.is_empty());
    }

    #[tokio::test]
    async fn test_multiple_shards_add_remove() {
        _test_multiple_shards_add_remove(mock_cdc_file_name(false)).await;
        _test_multiple_shards_add_remove(mock_cdc_file_name(true)).await;
    }

    #[tokio::test]
    async fn test_reader() {
        let backend_storage = test_create_backend_storage().await.unwrap();

        let files = backend_storage
            .list(TEST_S3_BUCKET, TEST_TABLE_PREFIX.as_str(), None)
            .await
            .unwrap()
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        assert!(files.len() >= 3);
        let first_file = files.get(files.len() - 3).unwrap().as_str();
        let last_file = files.last().unwrap().as_str();
        let prefix = TEST_TABLE_PREFIX.to_string();

        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: prefix,
            last_file: CDCFileName::empty(),
        };

        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let mut reader = CDCFileReader::new(
            backend_storage,
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        reader.start();
        // Add shard 1 last_file = first_file and process all files.
        let shard1 = mock_shard_meta(1, first_file);
        let (tx1, mut rx1) = mpsc::channel(100);
        reader.add_shard(mock_reader_shard_info(&shard1, tx1, index_meta.clone()));
        while let Some(entry) = rx1.recv().await {
            match entry {
                CDCLogEntry::CDCData(_) => {}
                CDCLogEntry::Filename(fname) => {
                    if fname.to_string() == last_file {
                        break;
                    }
                }
            }
        }
        assert_eq!(
            reader.get_shard_last_file(1).unwrap().to_string(),
            last_file.to_string()
        );

        // Add shard 2 last_file = first_file and verify it receives all filenames.
        let shard2 = mock_shard_meta(2, first_file);
        let (tx2, mut rx2) = mpsc::channel(100);
        reader.add_shard(mock_reader_shard_info(&shard2, tx2, index_meta.clone()));
        let mut receive_file_count = 0;
        while let Some(entry) = rx2.recv().await {
            match entry {
                CDCLogEntry::CDCData(_) => {}
                CDCLogEntry::Filename(fname) => {
                    receive_file_count += 1;
                    if fname.to_string() == last_file {
                        break;
                    }
                }
            }
        }
        assert_eq!(receive_file_count, 2);

        // remove shard 1 and 2
        reader.remove_shard(1);
        reader.remove_shard(2);

        // Add shard 3 last_file = files[n-2], should process only last file.
        let shard3 = mock_shard_meta(3, files.get(files.len() - 2).unwrap().as_str());
        let (tx3, mut rx3) = mpsc::channel(100);
        reader.add_shard(mock_reader_shard_info(&shard3, tx3, index_meta.clone()));
        while let Some(entry) = rx3.recv().await {
            match entry {
                CDCLogEntry::CDCData(_) => {}
                CDCLogEntry::Filename(fname) => {
                    assert_eq!(fname.to_string(), last_file);
                    break;
                }
            }
        }
        assert_eq!(
            reader.get_shard_last_file(3).unwrap().to_string(),
            last_file.to_string()
        );

        // Add shard 4 last_file = first_file, shard 5 last_file = last_file.
        // Shard 4 should process 2 files, shard 5 should process no file.
        let shard4 = mock_shard_meta(4, first_file);
        let (tx4, mut rx4) = mpsc::channel(100);
        reader.add_shard(mock_reader_shard_info(&shard4, tx4, index_meta.clone()));
        let shard5 = mock_shard_meta(5, last_file);
        let (tx5, rx5) = mpsc::channel(100);
        reader.add_shard(mock_reader_shard_info(&shard5, tx5, index_meta.clone()));
        let mut receive_file_count4 = 0;
        while let Some(entry) = rx4.recv().await {
            match entry {
                CDCLogEntry::CDCData(_) => {}
                CDCLogEntry::Filename(fname) => {
                    receive_file_count4 += 1;
                    if fname.to_string() == last_file {
                        break;
                    } else {
                        // shard 3's last_file should not change.
                        assert_eq!(
                            reader.get_shard_last_file(3).unwrap().to_string(),
                            last_file.to_string()
                        );
                    }
                }
            }
        }
        assert_eq!(receive_file_count4, 2);
        assert!(rx5.is_empty());

        // remove shard 3, 4 and 5
        reader.remove_shard(3);
        reader.remove_shard(4);
        reader.remove_shard(5);

        reader.stop().await;
    }

    #[tokio::test]
    async fn test_empty_poll_backoff_grows_and_resets() {
        let base_delay = Duration::from_millis(20);
        let mut backoff = EmptyPollBackoff::new(base_delay, Duration::from_millis(320));

        assert_eq!(backoff.current_delay(), base_delay);
        assert_eq!(backoff.on_empty_poll(), Duration::from_millis(40));
        assert_eq!(backoff.current_delay(), Duration::from_millis(40));
        assert_eq!(backoff.on_empty_poll(), Duration::from_millis(80));
        assert_eq!(backoff.on_empty_poll(), Duration::from_millis(160));

        for _ in 0..8 {
            backoff.on_empty_poll();
        }
        assert_eq!(backoff.current_delay(), Duration::from_millis(320));

        backoff.reset();
        assert_eq!(backoff.current_delay(), base_delay);
        assert_eq!(backoff.on_empty_poll(), Duration::from_millis(40));
    }

    #[tokio::test]
    async fn test_reader_does_not_overlap_list_requests() {
        let backend_storage =
            Arc::new(MockBackendStorage::new().with_blocking_list(HashMap::new()));
        let list_dispatcher = Arc::new(CDCListDispatcher::new(backend_storage.clone(), 4));
        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: "mock/table/123".to_string(),
            last_file: CDCFileName::empty(),
        };
        let mut reader = CDCFileReader::new_with_dispatcher(
            backend_storage.clone(),
            &cdc_checkpoint,
            Duration::from_millis(20),
            Duration::from_millis(320),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
            list_dispatcher,
        );

        reader.start();
        backend_storage.wait_for_started_calls(1).await;
        tokio::time::sleep(Duration::from_millis(80)).await;
        assert_eq!(backend_storage.started_list_calls(), 1);
        assert_eq!(backend_storage.max_active_list_calls(), 1);

        backend_storage.release_calls(1);
        backend_storage.wait_for_started_calls(2).await;
        assert_eq!(backend_storage.max_active_list_calls(), 1);

        backend_storage.release_calls(1);
        reader.stop().await;
    }

    #[tokio::test]
    async fn test_multiple_readers_share_dispatcher_and_process_files() {
        let backend_storage = Arc::new(MockBackendStorage::new());
        let list_dispatcher = Arc::new(CDCListDispatcher::new(backend_storage.clone(), 1));
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let mut readers = Vec::new();
        let mut receivers = Vec::new();
        let mut shard_ids = Vec::new();

        for idx in 0..4_u64 {
            let table_prefix = format!("mock/cdc/{}", 1000 + idx);
            let cdc_checkpoint = CDCCheckpoint {
                bucket: TEST_S3_BUCKET.to_string(),
                table_prefix: table_prefix.clone(),
                last_file: CDCFileName::empty(),
            };
            let filename = format!(
                "{}/458492412002304040/CDC00000000000000000001.json",
                table_prefix
            );
            backend_storage.push_list_response(
                TEST_S3_BUCKET,
                &format!("{table_prefix}/"),
                None,
                Ok(vec![filename.clone()]),
            );
            backend_storage.push_object_response(
                TEST_S3_BUCKET,
                &filename,
                Ok(format!(
                    "{}\n",
                    mock_insert_cdc_log_line(200 + idx, 1000 + idx, &format!("body_{idx}"))
                )
                .into_bytes()),
            );
            backend_storage.push_list_response(
                TEST_S3_BUCKET,
                &format!("{table_prefix}/"),
                Some(filename.as_str()),
                Ok(Vec::new()),
            );

            let mut reader = CDCFileReader::new_with_dispatcher(
                backend_storage.clone(),
                &cdc_checkpoint,
                Duration::from_millis(25),
                Duration::from_millis(400),
                TEST_CDC_PREFETCH_WINDOW,
                TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
                list_dispatcher.clone(),
            );
            reader.start();

            let shard_id = idx + 1;
            let shard_meta = mock_shard_meta_with_checkpoint(shard_id, cdc_checkpoint.clone());
            let (tx, rx) = mpsc::channel(8);
            reader.add_shard(mock_reader_shard_info(&shard_meta, tx, index_meta.clone()));
            readers.push((reader, filename));
            receivers.push(rx);
            shard_ids.push(shard_id);
        }

        for (idx, rx) in receivers.iter_mut().enumerate() {
            let expected_filename = readers[idx].1.clone();
            timeout(Duration::from_secs(5), async {
                while let Some(entry) = rx.recv().await {
                    if let CDCLogEntry::Filename(filename) = entry {
                        if filename.to_string() == expected_filename {
                            return;
                        }
                    }
                }
            })
            .await
            .unwrap();
        }

        for ((reader, filename), shard_id) in readers.iter().zip(shard_ids.iter()) {
            assert_eq!(
                reader.get_shard_last_file(*shard_id).unwrap().to_string(),
                *filename
            );
        }

        for (reader, _) in readers {
            reader.stop().await;
        }
    }

    #[tokio::test]
    async fn test_process_file_with_all_receivers_dropped() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let mut files = backend_storage
            .list(TEST_S3_BUCKET, TEST_TABLE_PREFIX.as_str(), None)
            .await
            .unwrap()
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        files.sort();
        let filename = CDCFileName::parse(files.first().unwrap()).unwrap();
        let prev_file = CDCFileName::parse("").unwrap();

        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: TEST_TABLE_PREFIX.to_string(),
            last_file: CDCFileName::empty(),
        };
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let reader = CDCFileReader::new(
            backend_storage,
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        let shard = mock_shard_meta(1, "");
        let (tx, rx) = mpsc::channel(8);
        drop(rx);
        reader.add_shard(mock_reader_shard_info(&shard, tx, index_meta));
        let filename_str = filename.to_string();
        let file_content = reader.file_cache.read_file(&filename_str).await.unwrap();

        CDCFileReader::process_file(file_content, &filename, &prev_file, &reader.state)
            .await
            .unwrap();

        let shard_info = reader.get_shard_info(1).unwrap();
        assert!(shard_info.is_closed());
        assert_eq!(shard_info.get_last_file(), prev_file);
    }

    #[tokio::test]
    async fn test_fill_prefetch_tasks_respects_prefetch_window() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let mut unread_files = backend_storage
            .list(TEST_S3_BUCKET, TEST_TABLE_PREFIX.as_str(), None)
            .await
            .unwrap()
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        unread_files.sort_unstable();
        assert!(unread_files.len() >= 3);

        let file_cache = Arc::new(CDCFileCache::new(
            backend_storage,
            TEST_S3_BUCKET,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        ));
        let mut inflight_prefetch = VecDeque::new();
        inflight_prefetch.push_back((
            unread_files[0].clone(),
            tokio::spawn(async {
                Ok::<FileContent, anyhow::Error>(FileContent {
                    data: String::new(),
                    stats: crate::writer::cdc_file_cache::FileContentStats::default(),
                })
            }),
        ));
        let mut next_prefetch_idx = 1;
        CDCFileReader::fill_prefetch_tasks(
            &file_cache,
            &unread_files[..3],
            &mut next_prefetch_idx,
            2,
            &mut inflight_prefetch,
        );

        assert_eq!(next_prefetch_idx, 2);
        assert_eq!(inflight_prefetch.len(), 2);
        assert_eq!(
            inflight_prefetch
                .front()
                .map(|(filename, _)| filename.as_str()),
            Some(unread_files[0].as_str())
        );
        assert_eq!(
            inflight_prefetch
                .back()
                .map(|(filename, _)| filename.as_str()),
            Some(unread_files[1].as_str())
        );

        CDCFileReader::abort_prefetch_tasks(&mut inflight_prefetch);
    }

    #[tokio::test]
    async fn test_process_file_uses_prefetched_content_without_reread() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let table_prefix = format!(
            "isolated/test/prefetched_content/{}/{}",
            rand::random_range(1_u64..=u64::MAX),
            TEST_INDEX_KEY.table_id
        );
        let filename_str = format!(
            "{}/458492412002304040/CDC00000000000000000001.json",
            table_prefix
        );
        backend_storage
            .put_data(
                TEST_S3_BUCKET,
                &filename_str,
                format!(
                    "{}\n",
                    mock_insert_cdc_log_line(200, 1000, "prefetched_body")
                )
                .into_bytes(),
            )
            .await
            .unwrap();
        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: table_prefix.clone(),
            last_file: CDCFileName::empty(),
        };
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let reader = CDCFileReader::new(
            backend_storage.clone(),
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );
        let shard = mock_shard_meta(1, "");
        let (tx, mut rx) = mpsc::channel(8);
        reader.add_shard(mock_reader_shard_info(&shard, tx, index_meta));

        let prefetched =
            CDCFileReader::spawn_prefetch_task(reader.file_cache.clone(), &filename_str)
                .await
                .unwrap()
                .unwrap();
        backend_storage
            .delete_objects(TEST_S3_BUCKET, std::slice::from_ref(&filename_str))
            .await
            .unwrap();

        let filename = CDCFileName::parse(&filename_str).unwrap();
        let prev_file = CDCFileName::parse("").unwrap();
        CDCFileReader::process_file(prefetched, &filename, &prev_file, &reader.state)
            .await
            .unwrap();

        let (cdc_data_count, filenames) = drain_receiver(&mut rx);
        assert_eq!(cdc_data_count, 1);
        assert_eq!(filenames, vec![filename.to_string()]);
        assert_eq!(reader.get_shard_last_file(1).unwrap(), filename);
    }

    #[tokio::test]
    async fn test_process_unread_files_keeps_poll_prev_file_snapshot() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let table_prefix = format!(
            "isolated/test/poll_prev_file_snapshot/{}/{}",
            rand::random_range(1_u64..=u64::MAX),
            TEST_INDEX_KEY.table_id
        );
        let listed_prev_file_str = format!(
            "{}/458492412002304040/CDC00000000000000000001.json",
            table_prefix
        );
        let unread_file_str = format!(
            "{}/458492412002304040/CDC00000000000000000002.json",
            table_prefix
        );
        backend_storage
            .put_data(
                TEST_S3_BUCKET,
                &unread_file_str,
                format!("{}\n", mock_insert_cdc_log_line(200, 1000, "lagging_shard")).into_bytes(),
            )
            .await
            .unwrap();

        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix,
            last_file: CDCFileName::parse(&listed_prev_file_str).unwrap(),
        };
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let reader = CDCFileReader::new(
            backend_storage.clone(),
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        // Simulate add_shard() rewinding shared state after list_unread_files()
        // already used listed_prev_file_str for this poll round.
        let lagging_shard = mock_shard_meta(1, "");
        let (tx, mut rx) = mpsc::channel(8);
        reader.add_shard(mock_reader_shard_info(&lagging_shard, tx, index_meta));

        let listed_prev_file = CDCFileName::parse(&listed_prev_file_str).unwrap();
        let rewind_file = CDCFileName::parse("").unwrap();
        assert_eq!(reader.state.get_last_file(), rewind_file);

        CDCFileReader::process_unread_files(
            &reader.file_cache,
            vec![unread_file_str.clone()],
            listed_prev_file,
            &reader.state,
            &CancellationToken::new(),
            TEST_CDC_PREFETCH_WINDOW,
        )
        .await
        .unwrap();

        let (cdc_data_count, filenames) = drain_receiver(&mut rx);
        assert_eq!(cdc_data_count, 0);
        assert!(filenames.is_empty());
        assert_eq!(reader.get_shard_last_file(1).unwrap(), rewind_file);
        assert_eq!(reader.state.get_last_file(), rewind_file);

        backend_storage
            .delete_objects(TEST_S3_BUCKET, &[unread_file_str])
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_reader_recovers_after_gap_retry() {
        let backend_storage = Arc::new(MockBackendStorage::new());
        let table_prefix = "isolated/test/retry_gap/123".to_string();
        let last_file = format!(
            "{}/458492412002304040/CDC00000000000000017429.json",
            table_prefix
        );
        let file_17430 = format!(
            "{}/458492412002304040/CDC00000000000000017430.json",
            table_prefix
        );
        let file_17431 = format!(
            "{}/458492412002304040/CDC00000000000000017431.json",
            table_prefix
        );
        let file_17432 = format!(
            "{}/458492412002304040/CDC00000000000000017432.json",
            table_prefix
        );
        let list_prefix = format!("{table_prefix}/");
        backend_storage.push_list_response(
            TEST_S3_BUCKET,
            &list_prefix,
            Some(last_file.as_str()),
            Ok(vec![file_17430.clone(), file_17432.clone()]),
        );
        backend_storage.push_list_response(
            TEST_S3_BUCKET,
            &list_prefix,
            Some(last_file.as_str()),
            Ok(vec![
                file_17430.clone(),
                file_17431.clone(),
                file_17432.clone(),
            ]),
        );
        backend_storage.push_object_response(
            TEST_S3_BUCKET,
            &file_17430,
            Ok(format!("{}\n", mock_insert_cdc_log_line(200, 1000, "body_17430")).into_bytes()),
        );
        backend_storage.push_object_response(
            TEST_S3_BUCKET,
            &file_17431,
            Ok(format!("{}\n", mock_insert_cdc_log_line(201, 1001, "body_17431")).into_bytes()),
        );
        backend_storage.push_object_response(
            TEST_S3_BUCKET,
            &file_17432,
            Ok(format!("{}\n", mock_insert_cdc_log_line(202, 1002, "body_17432")).into_bytes()),
        );

        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: table_prefix.clone(),
            last_file: CDCFileName::parse(&last_file).unwrap(),
        };
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let reader = CDCFileReader::new(
            backend_storage.clone(),
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        let shard = mock_shard_meta(1, &last_file);
        let (tx, mut rx) = mpsc::channel(16);
        reader.add_shard(mock_reader_shard_info(&shard, tx, index_meta));
        let table_prefix = CDCPrefix::new(table_prefix);
        let poller = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET);

        let err = poller
            .list_unread_files_for_table(&CDCFileName::parse(&last_file).unwrap(), &table_prefix)
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("Detected CDC file hole within directory")
        );
        let (cdc_data_count, filenames) = drain_receiver(&mut rx);
        assert_eq!(cdc_data_count, 0);
        assert!(filenames.is_empty());
        assert_eq!(reader.state.get_last_file().to_string(), last_file);
        assert_eq!(
            reader.get_shard_last_file(1).unwrap().to_string(),
            last_file
        );

        let unread_files = poller
            .list_unread_files_for_table(&CDCFileName::parse(&last_file).unwrap(), &table_prefix)
            .await
            .unwrap();
        let poll_prev_file = reader.state.get_last_file();
        CDCFileReader::process_unread_files(
            &reader.file_cache,
            unread_files,
            poll_prev_file,
            &reader.state,
            &CancellationToken::new(),
            TEST_CDC_PREFETCH_WINDOW,
        )
        .await
        .unwrap();

        let (cdc_data_count, filenames) = drain_receiver(&mut rx);
        assert_eq!(cdc_data_count, 3);
        assert_eq!(filenames, vec![file_17430, file_17431, file_17432.clone()]);
        assert_eq!(
            reader.get_shard_last_file(1).unwrap().to_string(),
            file_17432
        );
        assert_eq!(reader.state.get_last_file().to_string(), file_17432);
    }

    #[tokio::test]
    async fn test_process_file_error_contains_line_context() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let table_prefix = format!(
            "isolated/test/error_context/{}/{}",
            rand::random_range(1_u64..=u64::MAX),
            TEST_INDEX_KEY.table_id
        );
        let version_not_found_count_before = get_cdc_event_count("version_not_found");
        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: table_prefix.clone(),
            last_file: CDCFileName::empty(),
        };
        let index_meta = mock_custom_hdfs_logs_index_meta(TEST_INDEX_KEY.clone(), &[4]).unwrap();
        let reader = CDCFileReader::new(
            backend_storage.clone(),
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        let shard = mock_shard_meta(1, "");
        let (tx, _rx) = mpsc::channel(8);
        reader.add_shard(mock_reader_shard_info(&shard, tx, index_meta));

        let line = r#"{"id":0,"database":"test","table":"hdfs_15w","pkNames":["id","tenant_id"],"isDdl":false,"type":"UPDATE","es":1749009755276,"ts":1749009755613,"sql":"","sqlType":{"id":-5,"timestamp":-5,"severity_text":12,"body":2005,"tenant_id":4},"mysqlType":{"severity_text":"varchar","body":"text","tenant_id":"int","id":"bigint","timestamp":"bigint"},"old":[{"id":"1000","timestamp":"1460530013","severity_text":"INFO","body":"old body","tenant_id":"57"}],"data":[{"id":"1000","timestamp":"1460530013","severity_text":"INFO","body":"new body","tenant_id":"58"}],"_tidb":{"rowkey":"dIAAAAAAAAB7X3KAAAAAAAAD6Q=="}}"#;
        let filename_str = format!(
            "{}/458492412002304040/CDC00000000000000000001.json",
            table_prefix.as_str()
        );
        backend_storage
            .put_data(
                TEST_S3_BUCKET,
                &filename_str,
                format!("{line}\n").into_bytes(),
            )
            .await
            .unwrap();

        let filename = CDCFileName::parse(&filename_str).unwrap();
        let prev_file = CDCFileName::parse("").unwrap();
        let file_content = reader.file_cache.read_file(&filename_str).await.unwrap();
        let err = CDCFileReader::process_file(file_content, &filename, &prev_file, &reader.state)
            .await
            .unwrap_err();
        let err_msg = format!("{err:#}");

        assert!(err_msg.contains("Version not found"));
        assert!(err_msg.contains("line_num=0"));
        assert!(err_msg.contains(&format!("filename={}", filename)));
        assert!(err_msg.contains(&format!("line={}", line)));
        assert!(get_cdc_event_count("version_not_found") > version_not_found_count_before);

        backend_storage
            .delete_objects(TEST_S3_BUCKET, &[filename_str])
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_process_file_filters_events_before_created_ts() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let table_prefix = format!(
            "isolated/test/created_ts/{}/{}",
            rand::random_range(1_u64..=u64::MAX),
            TEST_INDEX_KEY.table_id
        );
        let filename_str = format!(
            "{}/458492412002304040/CDC00000000000000000001.json",
            table_prefix
        );
        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: table_prefix.clone(),
            last_file: CDCFileName::empty(),
        };
        let reader = CDCFileReader::new(
            backend_storage.clone(),
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );

        let shard1 = mock_shard_meta(1, "");
        let (tx1, mut rx1) = mpsc::channel(8);
        let index_meta1 = mock_hdfs_logs_index_meta_with_created_ts(TEST_INDEX_KEY.clone(), 200);
        reader.add_shard(mock_reader_shard_info(&shard1, tx1, index_meta1));

        let shard2 = mock_shard_meta(2, "");
        let (tx2, mut rx2) = mpsc::channel(8);
        let index_meta2 = mock_hdfs_logs_index_meta_with_created_ts(
            IndexKey::new(1, TEST_INDEX_KEY.table_id, 4),
            250,
        );
        reader.add_shard(mock_reader_shard_info(&shard2, tx2, index_meta2));

        let shard3 = mock_shard_meta(3, "");
        let (tx3, mut rx3) = mpsc::channel(8);
        let index_meta3 = mock_hdfs_logs_index_meta_with_created_ts(
            IndexKey::new(1, TEST_INDEX_KEY.table_id, 5),
            400,
        );
        reader.add_shard(mock_reader_shard_info(&shard3, tx3, index_meta3));

        let file_content = [
            mock_insert_cdc_log_line(100, 1000, "body_100"),
            mock_insert_cdc_log_line(200, 1001, "body_200"),
            mock_insert_cdc_log_line(300, 1002, "body_300"),
        ]
        .join("\n")
            + "\n";
        backend_storage
            .put_data(TEST_S3_BUCKET, &filename_str, file_content.into_bytes())
            .await
            .unwrap();

        let filename = CDCFileName::parse(&filename_str).unwrap();
        let prev_file = CDCFileName::parse("").unwrap();
        let file_content = reader.file_cache.read_file(&filename_str).await.unwrap();
        CDCFileReader::process_file(file_content, &filename, &prev_file, &reader.state)
            .await
            .unwrap();

        let (cdc_data_count1, filenames1) = drain_receiver(&mut rx1);
        assert_eq!(cdc_data_count1, 2);
        assert_eq!(filenames1, vec![filename.to_string()]);
        assert_eq!(reader.get_shard_last_file(1).unwrap(), filename);

        let (cdc_data_count2, filenames2) = drain_receiver(&mut rx2);
        assert_eq!(cdc_data_count2, 1);
        assert_eq!(filenames2, vec![filename.to_string()]);
        assert_eq!(reader.get_shard_last_file(2).unwrap(), filename);

        let (cdc_data_count3, filenames3) = drain_receiver(&mut rx3);
        assert_eq!(cdc_data_count3, 0);
        assert_eq!(filenames3, vec![filename.to_string()]);
        assert_eq!(reader.get_shard_last_file(3).unwrap(), filename);

        backend_storage
            .delete_objects(TEST_S3_BUCKET, &[filename_str])
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_stop_aborts_background_task() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let cdc_checkpoint = CDCCheckpoint {
            bucket: TEST_S3_BUCKET.to_string(),
            table_prefix: TEST_TABLE_PREFIX.to_string(),
            last_file: CDCFileName::empty(),
        };
        let mut reader = CDCFileReader::new(
            backend_storage,
            &cdc_checkpoint,
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );
        reader.background_handle = Some(tokio::spawn(async {
            std::future::pending::<Result<()>>().await
        }));

        tokio::time::timeout(Duration::from_secs(2), reader.stop())
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_reader_for_partition_table() {
        init_logger_for_ut();

        let backend_storage = test_create_backend_storage().await.unwrap();
        let prefix_437 = format!("{}/437", TEST_S3_PREFIX.as_str());
        let prefix_438 = format!("{}/438", TEST_S3_PREFIX.as_str());
        let prefix_439 = format!("{}/439", TEST_S3_PREFIX.as_str());
        let prefix_440 = format!("{}/440", TEST_S3_PREFIX.as_str());

        // Test files list as follows:
        // ----- partition table physical table id: 437
        // - {prefix}/463862231227695130/437/CDC00000000000000000001.json
        // - {prefix}/463862231227695130/437/CDC00000000000000000002.json
        // - {prefix}/463862231227695130/437/CDC00000000000000000003.json
        // - {prefix}/463862231227695130/437/CDC00000000000000000004.json
        // ----- partition table physical table id: 438
        // - {prefix}/463862231227695130/438/CDC00000000000000000001.json
        // - {prefix}/463862231227695130/438/CDC00000000000000000002.json
        // - {prefix}/463862231227695130/438/CDC00000000000000000003.json
        // - {prefix}/463862231227695130/438/CDC00000000000000000004.json
        // - {prefix}/463862231227695130/438/CDC00000000000000000005.json
        // ----- partition table physical table id: 439
        // - {prefix}/463862231227695130/439/CDC00000000000000000001.json
        // - {prefix}/463862231227695130/439/CDC00000000000000000002.json
        // - {prefix}/463862231227695130/439/CDC00000000000000000003.json
        // ----- partition table physical table id: 440
        // - {prefix}/463862231227695130/440/CDC00000000000000000001.json
        // - {prefix}/463862231227695130/440/CDC00000000000000000002.json
        // - {prefix}/463862231227695130/440/CDC00000000000000000003.json
        let mut files_437 = backend_storage
            .list(TEST_S3_BUCKET_PARTITION, &prefix_437, None)
            .await
            .unwrap()
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        files_437.sort();
        assert_eq!(files_437.len(), 4);

        let mut files_438 = backend_storage
            .list(TEST_S3_BUCKET_PARTITION, &prefix_438, None)
            .await
            .unwrap()
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        files_438.sort();
        assert_eq!(files_438.len(), 5);

        let mut files_439 = backend_storage
            .list(TEST_S3_BUCKET_PARTITION, &prefix_439, None)
            .await
            .unwrap()
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        files_439.sort();
        assert_eq!(files_439.len(), 3);

        let mut files_440 = backend_storage
            .list(TEST_S3_BUCKET_PARTITION, &prefix_440, None)
            .await
            .unwrap()
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        files_440.sort();
        assert_eq!(files_440.len(), 3);

        // Add shard 1, last_file = "", and process all files.
        // Locate at partition table physical table id: 437
        let shard1 = mock_shard_meta(1, "");
        let index_meta1 =
            mock_custom_employees_partition_index_meta(IndexKey::new(0, 437, 1)).unwrap();
        let mut reader1 = CDCFileReader::new(
            backend_storage.clone(),
            &CDCCheckpoint {
                bucket: TEST_S3_BUCKET_PARTITION.to_string(),
                table_prefix: prefix_437.clone(),
                last_file: CDCFileName::empty(),
            },
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );
        reader1.start();
        let (tx1, mut rx1) = mpsc::channel(100);
        reader1.add_shard(mock_reader_shard_info(&shard1, tx1, index_meta1));

        let mut receive_file_count = 0;
        let partition_last_file = files_437.last().unwrap().clone();
        while let Some(entry) = rx1.recv().await {
            match entry {
                CDCLogEntry::CDCData(_) => {}
                CDCLogEntry::Filename(fname) => {
                    receive_file_count += 1;
                    if fname.to_string() == partition_last_file {
                        break;
                    }
                }
            }
        }
        assert_eq!(
            reader1.get_shard_last_file(1).unwrap().to_string(),
            partition_last_file
        );
        assert_eq!(receive_file_count, 4);

        // Add shard 2, last_file = first_file, and process 3 files.
        // Locate at partition table physical table id: 438
        let first_file = files_438[1].as_str();
        let shard2 = mock_shard_meta(2, first_file);
        let index_meta2 =
            mock_custom_employees_partition_index_meta(IndexKey::new(0, 438, 1)).unwrap();
        let mut reader2 = CDCFileReader::new(
            backend_storage.clone(),
            &CDCCheckpoint {
                bucket: TEST_S3_BUCKET_PARTITION.to_string(),
                table_prefix: prefix_438.clone(),
                last_file: CDCFileName::empty(),
            },
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );
        reader2.start();
        let (tx2, mut rx2) = mpsc::channel(100);
        reader2.add_shard(mock_reader_shard_info(&shard2, tx2, index_meta2));

        let mut receive_file_count = 0;
        let partition_last_file = files_438.last().unwrap().clone();
        while let Some(entry) = rx2.recv().await {
            match entry {
                CDCLogEntry::CDCData(_) => {}
                CDCLogEntry::Filename(fname) => {
                    receive_file_count += 1;
                    if fname.to_string() == partition_last_file {
                        break;
                    }
                }
            }
        }
        assert_eq!(
            reader2.get_shard_last_file(2).unwrap().to_string(),
            partition_last_file
        );
        assert_eq!(receive_file_count, 3);

        // Add shard 3, process remaining files.
        // Locate at partition table physical table id: 439
        let first_file = files_439[1].clone();
        let shard3 = mock_shard_meta(3, &first_file);
        let index_meta3 =
            mock_custom_employees_partition_index_meta(IndexKey::new(0, 439, 1)).unwrap();
        let mut reader3 = CDCFileReader::new(
            backend_storage.clone(),
            &CDCCheckpoint {
                bucket: TEST_S3_BUCKET_PARTITION.to_string(),
                table_prefix: prefix_439.clone(),
                last_file: CDCFileName::empty(),
            },
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );
        reader3.start();
        let (tx3, mut rx3) = mpsc::channel(100);
        reader3.add_shard(mock_reader_shard_info(&shard3, tx3, index_meta3));

        let mut receive_file_count = 0;
        let partition_last_file = files_439.last().unwrap().clone();
        while let Some(entry) = rx3.recv().await {
            match entry {
                CDCLogEntry::CDCData(_) => {}
                CDCLogEntry::Filename(fname) => {
                    receive_file_count += 1;
                    if fname.to_string() == partition_last_file {
                        break;
                    }
                }
            }
        }
        assert_eq!(
            reader3.get_shard_last_file(3).unwrap().to_string(),
            partition_last_file
        );
        assert_eq!(receive_file_count, 1);

        // Add shard 4, process no file.
        // Locate at partition table physical table id: 440
        let first_file = files_440.last().unwrap().clone();
        let shard4 = mock_shard_meta(4, &first_file);
        let index_meta4 =
            mock_custom_employees_partition_index_meta(IndexKey::new(0, 440, 1)).unwrap();
        let mut reader4 = CDCFileReader::new(
            backend_storage.clone(),
            &CDCCheckpoint {
                bucket: TEST_S3_BUCKET_PARTITION.to_string(),
                table_prefix: prefix_440.clone(),
                last_file: CDCFileName::empty(),
            },
            Duration::from_secs(1),
            Duration::from_secs(1),
            TEST_CDC_PREFETCH_WINDOW,
            TEST_CDC_HEAD_OBJECT_LAST_MODIFIED_FALLBACK,
        );
        reader4.start();
        let (tx4, rx4) = mpsc::channel(100);
        reader4.add_shard(mock_reader_shard_info(&shard4, tx4, index_meta4.clone()));

        let partition_last_file = files_440.last().unwrap().clone();
        assert_eq!(
            reader4.get_shard_last_file(4).unwrap().to_string(),
            partition_last_file
        );
        assert!(rx4.is_empty());

        reader1.remove_shard(1);
        reader2.remove_shard(2);
        reader3.remove_shard(3);
        reader4.remove_shard(4);

        reader1.stop().await;
        reader2.stop().await;
        reader3.stop().await;
        reader4.stop().await;
    }
}