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

Code before:
            CDC_GRPC_ACCUMULATE_MESSAGE_BYTES.with_label_values(&["resolved_ts"]);

        let memory_quota = self.memory_quota.clone();
        let mut chunks = self.drain().ready_chunks(CDC_EVENT_MAX_COUNT);
        while let Some(events) = chunks.next().await {
            let mut bytes = 0;
            let mut batcher = EventBatcher::with_capacity(CDC_RESP_MAX_BATCH_COUNT);
            events.into_iter().for_each(|(e, size)| {
                bytes += size;
                batcher.push(e);
            });
            let (event_bytes, resolved_ts_bytes) = batcher.statistics();
            let resps = batcher.build();
            let resps_len = resps.len();
            // Events are about to be sent, free pending events memory counter.
            memory_quota.free(bytes as _);
            for (i, e) in resps.into_iter().enumerate() {
                // Buffer messages and flush them at once.
                let write_flags = WriteFlags::default().buffer_hint(i + 1 != resps_len);
                sink.feed((e, write_flags)).await?;
            }
            sink.flush().await?;
            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);

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:
            CDC_GRPC_ACCUMULATE_MESSAGE_BYTES.with_label_values(&["resolved_ts"]);

        let memory_quota = self.memory_quota.clone();
        let mut chunks = self.drain().ready_chunks(CDC_EVENT_MAX_COUNT);
        while let Some(events) = chunks.next().await {
            let mut bytes = 0;
            let mut batcher = EventBatcher::with_capacity(CDC_RESP_MAX_BATCH_COUNT);
            events.into_iter().for_each(|(e, size)| {
                bytes += size;
                batcher.push(e);
            });
            let (event_bytes, resolved_ts_bytes) = batcher.statistics();
            let resps = batcher.build();
            let resps_len = resps.len();
            // Events are about to be sent, free pending events memory counter.
            memory_quota.free(bytes as _);
            for (i, e) in resps.into_iter().enumerate() {
                // Buffer messages and flush them at once.
                let write_flags = WriteFlags::default().buffer_hint(i + 1 != resps_len);
                sink.feed((e, write_flags)).await?;
            }
            sink.flush().await?;
            #[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 {