Repository: tikv/tikv
File: components/cdc/src/channel.rs (Rust)

Code before:
            sleep_after_sink_flush().await;
            // Update last flush time if provided
            if let Some(time_tracker) = last_flush_time {
                use std::time::{SystemTime, UNIX_EPOCH};
                if let Ok(now) = SystemTime::now().duration_since(UNIX_EPOCH) {
                    time_tracker.store(now.as_secs(), std::sync::atomic::Ordering::Relaxed);
                }
            }  
            total_event_bytes.inc_by(event_bytes as u64);
            total_resolved_ts_bytes.inc_by(resolved_ts_bytes as u64);
        }
        Ok(())
    }
}


async fn sleep_after_sink_flush() {
    let should_sleep = || {
        fail::fail_point!("cdc_sleep_after_sink_flush", |_| true);
        false
    };
    if !should_sleep() {
        return;
    }
    info!("inside sleep_after_sink_flush failpoint, sleep 5 minutes!"); 
    use std::time::{Duration, Instant};    
    use tikv_util::timer::GLOBAL_TIMER_HANDLE;
    let dur = Duration::from_secs(60*5);
    let timer = GLOBAL_TIMER_HANDLE.delay(Instant::now() + dur);
    let _ = futures::compat::Compat01As03::new(timer).await;
}

impl Drop for Drain {
    fn drop(&mut self) {
        self.bounded_receiver.close();
        self.unbounded_receiver.close();
        let start = Instant::now();
        let mut total_bytes = 0;
        let mut drain = Box::pin(async move {
            let conn_id = self.conn_id;
            let memory_quota = self.memory_quota.clone();
            let mut drain = self.drain();
            while let Some((_, bytes)) = drain.next().await {
                total_bytes += bytes;
            }
            memory_quota.free(total_bytes);
            info!("drop Drain finished, free memory"; "conn_id" => ?conn_id,
                "freed_bytes" => total_bytes, "inuse_bytes" => memory_quota.in_use());
        });
        block_on(&mut drain);
        let takes = start.saturating_elapsed();

Review comment (suggestion):
It's better to place use statements at the top of the function or module rather than inside a block.

Code after:
            #[cfg(feature = "failpoints")]
            sleep_after_sink_flush().await;
            // Update last flush time if provided
            if let Some(time_tracker) = last_flush_time {
                time_tracker.store(std::time::Instant::now());
            }
            total_event_bytes.inc_by(event_bytes as u64);
            total_resolved_ts_bytes.inc_by(resolved_ts_bytes as u64);
        }
        Ok(())
    }
}

#[cfg(feature = "failpoints")]
async fn sleep_after_sink_flush() {
    let should_sleep = || {
        fail::fail_point!("cdc_sleep_after_sink_flush", |_| true);
        false
    };
    if !should_sleep() {
        return;
    }
    info!("inside sleep_after_sink_flush failpoint, sleep 30 seconds!");
    let dur = Duration::from_secs(30);
    let timer = tikv_util::timer::GLOBAL_TIMER_HANDLE.delay(std::time::Instant::now() + dur);
    let _ = futures::compat::Compat01As03::new(timer).await;
}

impl Drop for Drain {
    fn drop(&mut self) {
        self.bounded_receiver.close();
        self.unbounded_receiver.close();
        let start = Instant::now();
        let mut total_bytes = 0;
        let mut drain = Box::pin(async move {
            let conn_id = self.conn_id;
            let memory_quota = self.memory_quota.clone();
            let mut drain = self.drain();
            while let Some((_, bytes)) = drain.next().await {
                total_bytes += bytes;
            }
            memory_quota.free(total_bytes);
            info!("drop Drain finished, free memory"; "conn_id" => ?conn_id,
                "freed_bytes" => total_bytes, "inuse_bytes" => memory_quota.in_use());
        });
        block_on(&mut drain);
        let takes = start.saturating_elapsed();
        if takes >= Duration::from_millis(200) {
            warn!("drop Drain too slow"; "takes" => ?takes);
        }
    }