Repository: risingwavelabs/risingwave
File: src/meta/src/barrier/context/recovery.rs (Rust)

Code before:
            .load_all_actors_dynamic(database_id, worker_nodes)
            .await?;

        Ok(all_actor_infos
            .into_iter()
            .map(|(loaded_database_id, job_fragment_infos)| {
                if let Some(database_id) = database_id {
                    assert_eq!(database_id, loaded_database_id);
                }
                (
                    loaded_database_id,
                    job_fragment_infos
                        .into_iter()
                        .map(|(job_id, fragment_infos)| {
                            (
                                job_id,
                                fragment_infos
                                    .into_iter()
                                    .map(|(fragment_id, info)| (fragment_id as _, info))
                                    .collect(),
                            )
                        })
                        .collect(),
                )
            })
            .collect())
    }

    #[expect(clippy::type_complexity)]
    fn resolve_hummock_version_epochs(
        background_jobs: impl Iterator<Item = (JobId, &HashMap<FragmentId, InflightFragmentInfo>)>,
        version: &HummockVersion,
    ) -> MetaResult<(
        HashMap<TableId, u64>,
        HashMap<TableId, Vec<(Vec<u64>, u64)>>,
    )> {
        let table_committed_epoch: HashMap<_, _> = version
            .state_table_info
            .info()
            .iter()
            .map(|(table_id, info)| (*table_id, info.committed_epoch))
            .collect();
        let get_table_committed_epoch = |table_id| -> anyhow::Result<u64> {
            Ok(*table_committed_epoch
                .get(&table_id)
                .ok_or_else(|| anyhow!("cannot get committed epoch on table {}.", table_id))?)
        };
        let mut min_downstream_committed_epochs = HashMap::new();
        for (job_id, fragments) in background_jobs {
            let job_committed_epoch = {
                let mut table_id_iter =

Review comment (suggestion):
Using `.unwrap()` on `NonZeroUsize::new()` can cause a panic if `compute_node_parallelism()` returns 0. While there's validation at compute node startup, there's no guarantee this value is always non-zero at runtime. Consider handling this more gracefully with proper error handling or validation before creating the WorkerInfo, such as filtering out workers with zero parallelism or returning a descriptive error.

Code after:
            .catalog_controller
            .load_fragment_context(database_id)
            .await
    }

    /// Sync render stage: use loaded context and current workers to produce actor assignments.
    fn render_actor_assignments(
        &self,
        database_id: Option<DatabaseId>,
        loaded: &LoadedFragmentContext,
        worker_nodes: &ActiveStreamingWorkerNodes,
        adaptive_parallelism_strategy: AdaptiveParallelismStrategy,
    ) -> MetaResult<FragmentRenderMap> {
        if loaded.is_empty() {
            return Ok(HashMap::new());
        }

        let available_workers: BTreeMap<_, _> = worker_nodes
            .current()
            .values()
            .filter(|worker| worker.is_streaming_schedulable())
            .map(|worker| {
                (
                    worker.id,
                    WorkerInfo {
                        parallelism: NonZeroUsize::new(worker.compute_node_parallelism()).unwrap(),
                        resource_group: worker.resource_group(),
                    },
                )
            })
            .collect();

        let RenderedGraph { fragments, .. } = render_actor_assignments(
            self.metadata_manager
                .catalog_controller
                .env
                .actor_id_generator(),
            &available_workers,
            adaptive_parallelism_strategy,
            loaded,
        )?;

        if let Some(database_id) = database_id {
            for loaded_database_id in fragments.keys() {
                assert_eq!(*loaded_database_id, database_id);
            }
        }

        Ok(fragments)
    }