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

Code before:

            let (_, database_id) = self
                .metadata_manager
                .catalog_controller
                .try_abort_creating_streaming_job(id, true)
                .await?;

            if let Some(database_id) = database_id {
                self.barrier_scheduler
                    .run_command(database_id, cancel_command)
                    .await?;
            }

            tracing::info!(?id, "cancelled recovered streaming job");
            Ok(id)
        });
        let cancelled_recovered_ids = join_all(futures)
            .await
            .into_iter()
            .collect::<MetaResult<Vec<_>>>()
            .map_err(|e| {
                MetaError::from(anyhow::anyhow!(
                    "failed to cancel recovered streaming jobs: {}",
                    e
                ))
            })?;

        cancelled_ids.extend(cancelled_recovered_ids);
        Ok(cancelled_ids)
    }

    pub(crate) async fn reschedule_streaming_job(
        &self,
        job_id: JobId,
        policy: ReschedulePolicy,
        deferred: bool,
    ) -> MetaResult<()> {
        let _reschedule_job_lock = self.reschedule_lock_write_guard().await;

        let background_jobs = self
            .metadata_manager
            .list_background_creating_jobs()
            .await?;

        if !background_jobs.is_empty() {
            let related_jobs = self
                .scale_controller
                .resolve_related_no_shuffle_jobs(&background_jobs)
                .await?;

            if related_jobs.contains(&job_id) {

Review comment (suggestion):
The error wrapping at lines 706-711 is redundant. The `collect::<MetaResult<Vec<_>>>()` at line 705 already propagates the error as a `MetaError`. Wrapping it again with `map_err` and `anyhow::anyhow!` loses the original error context and converts it to a less structured error. Consider removing the `map_err` call and using just `.collect::<MetaResult<Vec<_>>>()?;` to preserve the original error information.
```suggestion
            .collect::<MetaResult<Vec<_>>>()?;
```

Code after:
            let (_, database_id) = self
                .metadata_manager
                .catalog_controller
                .try_abort_creating_streaming_job(id, true)
                .await?;

            if let Some(database_id) = database_id {
                self.barrier_scheduler
                    .run_command(database_id, cancel_command)
                    .await?;
            }

            tracing::info!(?id, "cancelled recovered streaming job");
            Ok(id)
        });
        let cancelled_recovered_ids = join_all(futures)
            .await
            .into_iter()
            .collect::<MetaResult<Vec<_>>>()?;

        cancelled_ids.extend(cancelled_recovered_ids);
        Ok(cancelled_ids)
    }

    pub(crate) async fn reschedule_streaming_job(
        &self,
        job_id: JobId,
        policy: ReschedulePolicy,
        deferred: bool,
    ) -> MetaResult<()> {
        let _reschedule_job_lock = self.reschedule_lock_write_guard().await;

        let background_jobs = self
            .metadata_manager
            .list_background_creating_jobs()
            .await?;

        if !background_jobs.is_empty() {
            let related_jobs = self
                .scale_controller
                .resolve_related_no_shuffle_jobs(&background_jobs)
                .await?;

            if related_jobs.contains(&job_id) {
                bail!(
                    "Cannot alter the job {} because the related job {:?} is currently being created",
                    job_id,
                    background_jobs,
                );
            }
        }