Repository: risingwavelabs/risingwave
File: src/meta/src/stream/stream_manager.rs (Rust)

Code before:

    async fn delete_job(&self, job_id: JobId) {
        let mut jobs = self.streaming_jobs.lock().await;
        jobs.remove(&job_id);
    }

    async fn cancel_jobs(
        &self,
        job_ids: Vec<JobId>,
    ) -> MetaResult<(HashMap<JobId, oneshot::Receiver<bool>>, Vec<JobId>)> {
        let mut jobs = self.streaming_jobs.lock().await;
        let mut receivers = HashMap::new();
        let mut background_job_ids = vec![];
        for job_id in job_ids {
            if let Some(job) = jobs.get_mut(&job_id) {
                if let Some(shutdown_tx) = job.shutdown_tx.take() {
                    let (tx, rx) = oneshot::channel();
                    match shutdown_tx.send(tx) {
                        Ok(()) => {
                            receivers.insert(job_id, rx);
                        }
                        Err(e) => {
                            return Err(anyhow::anyhow!(
                                "failed to send shutdown signal for streaming job {}: {:?}",
                                job_id,
                                e
                            )
                            .into());
                        }
                    }
                }
            } else {
                // If these job ids do not exist in streaming_jobs, they should be background creating jobs.
                background_job_ids.push(job_id);
            }
        }

        Ok((receivers, background_job_ids))
    }

    async fn check_job_exists(&self, job_id: JobId) -> bool {
        let jobs = self.streaming_jobs.lock().await;
        jobs.contains_key(&job_id)
    }
}

type CreatingStreamingJobInfoRef = Arc<CreatingStreamingJobInfo>;

#[derive(Debug, Clone)]
pub struct AutoRefreshSchemaSinkContext {
    pub tmp_sink_id: JobId,

Review comment (bug):
The error formatting `{:?}` on line 168 will print the `oneshot::Sender<oneshot::Sender<bool>>` that failed to send, not a meaningful error message. When `oneshot::Sender::send()` fails, it returns `Err(value)` where `value` is the data that couldn't be sent. Consider simplifying the error message to just describe the failure without trying to format the sender, e.g., `"failed to send shutdown signal for streaming job {}: receiver dropped"`, or omit the error formatting entirely.
```suggestion
                                "failed to send shutdown signal for streaming job {}: receiver dropped",
                                job_id
```

Code after:

    async fn delete_job(&self, job_id: JobId) {
        let mut jobs = self.streaming_jobs.lock().await;
        jobs.remove(&job_id);
    }

    async fn cancel_jobs(
        &self,
        job_ids: Vec<JobId>,
    ) -> MetaResult<(HashMap<JobId, oneshot::Receiver<bool>>, Vec<JobId>)> {
        let mut jobs = self.streaming_jobs.lock().await;
        let mut receivers = HashMap::new();
        let mut background_job_ids = vec![];
        for job_id in job_ids {
            if let Some(job) = jobs.get_mut(&job_id) {
                if let Some(shutdown_tx) = job.shutdown_tx.take() {
                    let (tx, rx) = oneshot::channel();
                    match shutdown_tx.send(tx) {
                        Ok(()) => {
                            receivers.insert(job_id, rx);
                        }
                        Err(_) => {
                            return Err(anyhow::anyhow!(
                                "failed to send shutdown signal for streaming job {}: receiver dropped",
                                job_id
                            )
                            .into());
                        }
                    }
                }
            } else {
                // If these job ids do not exist in streaming_jobs, they should be background creating jobs.
                background_job_ids.push(job_id);
            }
        }

        Ok((receivers, background_job_ids))
    }

    async fn check_job_exists(&self, job_id: JobId) -> bool {
        let jobs = self.streaming_jobs.lock().await;
        jobs.contains_key(&job_id)
    }
}

type CreatingStreamingJobInfoRef = Arc<CreatingStreamingJobInfo>;

#[derive(Debug, Clone)]
pub struct AutoRefreshSchemaSinkContext {
    pub tmp_sink_id: JobId,
    pub original_sink: PbSink,