Repository: tikv/tikv
File: components/cdc/tests/failpoints/test_endpoint.rs (Rust)

Code before:
}

#[test]
fn test_cdc_watchdog_idle_timeout() {
    let cluster = new_server_cluster(0, 1);
    cluster.pd_client.disable_default_operator();
    let mut suite = TestSuiteBuilder::new().cluster(cluster).build();
    let region = suite.cluster.get_region(b"");

    // Enable failpoints to control the watchdog behavior
    // cdc_idle_deregister_threshold will make the threshold 20 seconds instead of
    // 20 minutes cdc_sleep_after_sink_flush will make the sink sleep for 30
    // seconds after each flush
    fail::cfg("cdc_idle_deregister_threshold", "return(true)").unwrap();
    fail::cfg("cdc_sleep_after_sink_flush", "return(true)").unwrap(); // Remove the "1*" to make it trigger continuously

    // Create event feed connection
    let (mut req_tx, event_feed, _) = new_event_feed(suite.get_region_cdc_client(region.id));
    let mut req = suite.new_changedata_request(region.id);
    req.mut_header().set_ticdc_version("7.5.0".into());
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();

    // Wait for the connection to be established and initialized
    thread::sleep(Duration::from_millis(1000));

    eprintln!("Starting watchdog test - waiting for connection to be cancelled");

    // Wait for the watchdog to trigger and cancel the connection
    // The watchdog should trigger after 20 seconds due to
    // cdc_idle_deregister_threshold failpoint and cdc_sleep_after_sink_flush
    // failpoint will make the sink sleep for 30 seconds
    thread::sleep(Duration::from_secs(30));

    eprintln!("Finished waiting, now checking if connection was cancelled");

    // Try to detect if the connection was cancelled by watchdog
    // We can do this by trying to receive from the underlying receiver
    // If the connection is closed, recv_timeout should return an error
    let mut connection_cancelled = false;
    let start_time = Instant::now();

    // Try to detect connection closure for up to 5 seconds (shorter timeout for
    // testing)
    while start_time.elapsed() < Duration::from_secs(5) {
        // Get the underlying receiver
        let mut rx = event_feed.replace(None).unwrap();

        // Try to receive with a short timeout
        match recv_timeout(&mut rx, Duration::from_millis(100)) {
            Ok(Some(Ok(_))) => {
                // Still receiving data, connection is alive

Review comment (bug):
```suggestion
    debug!("Starting watchdog test - waiting for connection to be cancelled");
```

Or remove it.

Code after:
}

#[test]
fn test_cdc_watchdog_idle_timeout() {
    let cluster = new_server_cluster(0, 1);
    cluster.pd_client.disable_default_operator();
    let mut suite = TestSuiteBuilder::new().cluster(cluster).build();
    let region = suite.cluster.get_region(b"");

    // Enable failpoints to control the watchdog behavior
    // cdc_idle_deregister_threshold will make the threshold 20 seconds instead of
    // 20 minutes cdc_sleep_after_sink_flush will make the sink sleep for 30
    // seconds after each flush
    fail::cfg("cdc_idle_deregister_threshold", "return(true)").unwrap();
    fail::cfg("cdc_sleep_after_sink_flush", "return(true)").unwrap(); // Remove the "1*" to make it trigger continuously

    // Create event feed connection
    let (mut req_tx, event_feed, _) = new_event_feed(suite.get_region_cdc_client(region.id));
    let mut req = suite.new_changedata_request(region.id);
    req.mut_header().set_ticdc_version("7.5.0".into());
    block_on(req_tx.send((req, WriteFlags::default()))).unwrap();

    // Wait for the connection to be established and initialized
    thread::sleep(Duration::from_millis(1000));

    debug!("Starting watchdog test - waiting for connection to be cancelled");

    // Wait for the watchdog to trigger and cancel the connection
    // The watchdog should trigger after 5 seconds due to
    // cdc_idle_deregister_threshold failpoint and cdc_sleep_after_sink_flush
    // failpoint will make the sink sleep for 6 seconds
    thread::sleep(Duration::from_secs(6));

    debug!("Finished waiting, now checking if connection was cancelled");

    // Try to detect if the connection was cancelled by watchdog
    // We can do this by trying to receive from the underlying receiver
    // If the connection is closed, recv_timeout should return an error
    let mut connection_cancelled = false;
    let start_time = Instant::now();

    // Try to detect connection closure for up to 5 seconds (shorter timeout for
    // testing)
    while start_time.elapsed() < Duration::from_secs(5) {
        // Get the underlying receiver
        let mut rx = event_feed.replace(None).unwrap();

        // Try to receive with a short timeout
        match recv_timeout(&mut rx, Duration::from_millis(100)) {
            Ok(Some(Ok(_))) => {
                // Still receiving data, connection is alive