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

Code before:
        sink.enhance_batch(true);
        let conn_id = ConnId::new();
        let (event_sink, mut event_drain) =
            channel(conn_id, CDC_CHANNLE_CAPACITY, self.memory_quota.clone());
        let conn = Conn::new(conn_id, event_sink, ctx.peer());
        let mut explicit_features = vec![];

        if event_feed_v2 {
            let headers = match Self::parse_headers(&ctx) {
                Ok(headers) => headers,
                Err(e) => {
                    let peer = ctx.peer();
                    error!("cdc connection with bad headers"; "downstream" => ?peer, "headers" => &e);
                    ctx.spawn(async move {
                        let status = RpcStatus::with_message(RpcStatusCode::UNIMPLEMENTED, e);
                        if let Err(e) = sink.fail(status).await {
                            error!("cdc failed to send error"; "downstream" => ?peer, "error" => ?e);
                        }
                    });
                    return;
                }
            };
            explicit_features = headers.features;
        }
        info!("cdc connection created"; "downstream" => ctx.peer(),
         "features" => ?explicit_features, "conn_id" => ?conn_id);

        if let Err(e) = self.scheduler.schedule(Task::OpenConn { conn }) {
            let peer = ctx.peer();
            error!("cdc connection initiate failed"; "downstream" => ?peer, "error" => ?e);
            ctx.spawn(async move {
                let status = RpcStatus::with_message(RpcStatusCode::UNKNOWN, format!("{:?}", e));
                if let Err(e) = sink.fail(status).await {
                    error!("cdc failed to send error"; "downstream" => ?peer, "error" => ?e);
                }
            });
            return;
        }

        let peer = ctx.peer();
        let scheduler = self.scheduler.clone();
        let recv_req = async move {
            let mut stream = stream.map_err(|e| format!("{:?}", e));
            if let Some(request) = stream.try_next().await? {
                // Get version from the first request in the stream.
                let version = Self::parse_version_from_request_header(&request, &peer);
                Self::set_conn_version(&scheduler, conn_id, version, explicit_features)?;
                Self::handle_request(&scheduler, &peer, request, conn_id)?;
            }
            while let Some(request) = stream.try_next().await? {
                Self::handle_request(&scheduler, &peer, request, conn_id)?;

Review comment (suggestion):
set the feature as first field since downstream print the ip address, it's more important than the feature, which always print `multiplexing`

Code after:
        event_feed_v2: bool,
    ) {
        sink.enhance_batch(true);
        let conn_id = ConnId::new();
        let (event_sink, mut event_drain) =
            channel(conn_id, CDC_CHANNLE_CAPACITY, self.memory_quota.clone());
        let conn = Conn::new(conn_id, event_sink, ctx.peer());
        let mut explicit_features = vec![];

        if event_feed_v2 {
            let headers = match Self::parse_headers(&ctx) {
                Ok(headers) => headers,
                Err(e) => {
                    let peer = ctx.peer();
                    error!("cdc connection with bad headers"; "downstream" => ?peer, "headers" => &e);
                    ctx.spawn(async move {
                        let status = RpcStatus::with_message(RpcStatusCode::UNIMPLEMENTED, e);
                        if let Err(e) = sink.fail(status).await {
                            error!("cdc failed to send error"; "downstream" => ?peer, "error" => ?e);
                        }
                    });
                    return;
                }
            };
            explicit_features = headers.features;
        }
        info!("cdc connection created"; "downstream" => ctx.peer(),
         "conn_id" => ?conn_id,"features" => ?explicit_features);

        if let Err(e) = self.scheduler.schedule(Task::OpenConn { conn }) {
            let peer = ctx.peer();
            error!("cdc connection initiate failed"; "downstream" => ?peer, "error" => ?e);
            ctx.spawn(async move {
                let status = RpcStatus::with_message(RpcStatusCode::UNKNOWN, format!("{:?}", e));
                if let Err(e) = sink.fail(status).await {
                    error!("cdc failed to send error"; "downstream" => ?peer, "error" => ?e);
                }
            });
            return;
        }

        let peer = ctx.peer();
        let scheduler = self.scheduler.clone();
        let recv_req = async move {
            let mut stream = stream.map_err(|e| format!("{:?}", e));
            if let Some(request) = stream.try_next().await? {
                // Get version from the first request in the stream.
                let version = Self::parse_version_from_request_header(&request, &peer);
                Self::set_conn_version(&scheduler, conn_id, version, explicit_features)?;
                Self::handle_request(&scheduler, &peer, request, conn_id)?;
            }