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

Code before:
        Ok(features)
    }
}

/// Service implements the `ChangeData` service.
///
/// It's a front-end of the CDC service, schedules requests to the `Endpoint`.
#[derive(Clone)]
pub struct Service {
    scheduler: Scheduler<Task>,
    memory_quota: Arc<MemoryQuota>,
    pool: Arc<Runtime>,
}

impl Service {
    /// Create a ChangeData service.
    ///
    /// It requires a scheduler of an `Endpoint` in order to schedule tasks.
    pub fn new(scheduler: Scheduler<Task>, memory_quota: Arc<MemoryQuota>) -> Service {
        let pool = Arc::new(
            tokio::runtime::Builder::new_multi_thread()
                .worker_threads(1)
                .thread_name("cdc-watchdog")
                .enable_all()
                .build()
                .unwrap(),
        );

        Service {
            scheduler,
            memory_quota,
            pool,
        }
    }

    // Parse HTTP/2 headers. Only for `Self::event_feed_v2`.
    fn parse_headers(ctx: &RpcContext<'_>) -> Result<EventFeedHeaders, String> {
        let mut header = EventFeedHeaders::default();
        let metadata = ctx.request_headers();
        for i in 0..metadata.len() {
            let (key, value) = metadata.get(i).unwrap();
            if key == EventFeedHeaders::FEATURES_KEY {
                header.features = EventFeedHeaders::parse_features(value)?;
            }
        }
        Ok(header)
    }

    fn parse_version_from_request_header(
        request: &ChangeDataRequest,
        peer: &str,

Review comment (suggestion):
Please avoid creating new threads. You can use background workers https://github.com/tikv/tikv/blob/d2f1a3705b676d61f7917bbff15e5002d92b9ef5/components/server/src/server.rs#L376

Code after:
        if !unknowns.is_empty() {
            return Err(unknowns.join(","));
        }
        Ok(features)
    }
}

/// Service implements the `ChangeData` service.
///
/// It's a front-end of the CDC service, schedules requests to the `Endpoint`.
#[derive(Clone)]
pub struct Service {
    scheduler: Scheduler<Task>,
    memory_quota: Arc<MemoryQuota>,
    pool: Arc<Worker>,
}

impl Service {
    /// Create a ChangeData service.
    ///
    /// It requires a scheduler of an `Endpoint` in order to schedule tasks.
    pub fn new(
        scheduler: Scheduler<Task>,
        memory_quota: Arc<MemoryQuota>,
        pool: Arc<Worker>,
    ) -> Service {
        Service {
            scheduler,
            memory_quota,
            pool,
        }
    }

    // Parse HTTP/2 headers. Only for `Self::event_feed_v2`.
    fn parse_headers(ctx: &RpcContext<'_>) -> Result<EventFeedHeaders, String> {
        let mut header = EventFeedHeaders::default();
        let metadata = ctx.request_headers();
        for i in 0..metadata.len() {
            let (key, value) = metadata.get(i).unwrap();
            if key == EventFeedHeaders::FEATURES_KEY {
                header.features = EventFeedHeaders::parse_features(value)?;
            }
        }
        Ok(header)
    }

    fn parse_version_from_request_header(
        request: &ChangeDataRequest,
        peer: &str,
    ) -> semver::Version {
        let version_field = request.get_header().get_ticdc_version();