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

Code before:
        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,
    pub original_sink: PbSink,
    pub original_fragment: Fragment,
    pub new_schema: Vec<PbColumnCatalog>,

Review comment (bug):
When `shutdown_tx.send(tx)` fails for one job, the function returns immediately without attempting to cancel the remaining jobs in the `job_ids` list. This changes the behavior from the previous implementation which would log a warning and continue processing other jobs. Consider collecting all errors and either continuing to process remaining jobs or returning a more informative error that indicates which jobs failed and which succeeded.

Code after:
        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,
    pub original_fragment: Fragment,
    pub new_schema: Vec<PbColumnCatalog>,
    pub newly_add_fields: Vec<Field>,