Repository: tikv/tikv
File: components/backup-stream/src/router.rs (Rust)

Code before:
                    task_handler.set_flushing_status(false);
                }
            }
        }
        Ok(())
    }

    pub async fn on_events(&self, kv: ApplyEvents) -> Vec<(String, Result<()>)> {
        use futures::FutureExt;
        HANDLE_KV_HISTOGRAM.observe(kv.len() as _);
        let partitioned_events = kv.partition_by_range(&self.ranges.rl());
        let tasks = partitioned_events.into_iter().map(|(task, events)| {
            self.on_events_by_task(task.clone(), events)
                .map(move |r| (task, r))
        });
        futures::future::join_all(tasks).await
    }

    /// flush the specified task, once success, return the min resolved ts
    /// of this flush. returns `None` if failed.
    #[instrument(skip(self, cx))]
    pub async fn do_flush(&self, cx: FlushContext<'_>) -> Option<u64> {
        let task = self.tasks.get(cx.task_name);
        match task {
            Some(task_handler) => {
                let result = task_handler.do_flush(cx).await;
                // set false to flushing whether success or fail
                task_handler.set_flushing_status(false);

                if let Err(e) = result {
                    e.report("failed to flush task.");
                    warn!("backup steam do flush fail"; "err" => ?e);
                    if task_handler.flush_failure_count() > FLUSH_FAILURE_BECOME_FATAL_THRESHOLD {
                        // NOTE: Maybe we'd better record all errors and send them to the client?
                        try_send!(
                            self.scheduler,
                            Task::FatalError(
                                TaskSelector::ByName(cx.task_name.to_owned()),
                                Box::new(e)
                            )
                        );
                    }
                    return None;
                }
                // if succeed in flushing, update flush_time. Or retry do_flush immediately.
                task_handler.update_flush_time();
                result.ok().flatten()
            }
            _ => None,
        }
    }

Review comment (suggestion):
```suggestion
        let task_handler = match self.get_task_handler(cx.task_name).ok()?;
```

Code after:
                    task_handler.set_flushing_status(false);
                }
            }
        }
        Ok(())
    }

    pub async fn on_events(&self, kv: ApplyEvents) -> Vec<(String, Result<()>)> {
        use futures::FutureExt;
        HANDLE_KV_HISTOGRAM.observe(kv.len() as _);
        let partitioned_events = kv.partition_by_range(&self.ranges.rl());
        let tasks = partitioned_events.into_iter().map(|(task, events)| {
            self.on_events_by_task(task.clone(), events)
                .map(move |r| (task, r))
        });
        futures::future::join_all(tasks).await
    }

    /// flush the specified task, once success, return the min resolved ts
    /// of this flush. returns `None` if failed.
    #[instrument(skip(self, cx))]
    pub async fn do_flush(&self, cx: FlushContext<'_>) -> Option<u64> {
        let task_handler = match self.get_task_handler(cx.task_name) {
            Ok(h) => h,
            Err(_) => return None,
        };
        let result = task_handler.do_flush(cx).await;
        // set false to flushing whether success or fail
        task_handler.set_flushing_status(false);

        if let Err(e) = result {
            e.report("failed to flush task.");
            warn!("backup steam do flush fail"; "err" => ?e);
            if task_handler.flush_failure_count() > FLUSH_FAILURE_BECOME_FATAL_THRESHOLD {
                // NOTE: Maybe we'd better record all errors and send them to the client?
                try_send!(
                    self.scheduler,
                    Task::FatalError(TaskSelector::ByName(cx.task_name.to_owned()), Box::new(e))
                );
            }
            return None;
        }
        // if succeed in flushing, update flush_time. Or retry do_flush immediately.
        task_handler.update_flush_time();
        result.ok().flatten()
    }

    #[instrument(skip(self))]
    pub async fn update_global_checkpoint(
        &self,
        task_name: &str,