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

Code before:
                    }
                }
            }
        }
    }

    async fn apply_post_backfill_parallelism(&self, job_id: JobId) -> MetaResult<()> {
        let (target, backfill_parallelism) = self
            .metadata_manager
            .catalog_controller
            .get_job_parallelisms(job_id)
            .await?;

        match backfill_parallelism {
            Some(backfill_parallelism) if backfill_parallelism == target => {
                tracing::debug!(
                    job_id = %job_id,
                    ?backfill_parallelism,
                    ?target,
                    "backfill parallelism equals job parallelism, skip reschedule"
                );
                return Ok(());
            }
            Some(_) => {
                // proceed to restore to target parallelism
            }
            None => {
                tracing::debug!(
                    job_id = %job_id,
                    ?target,
                    "no backfill parallelism configured, skip post-backfill reschedule"
                );
                return Ok(());
            }
        }

        tracing::info!(
            job_id = %job_id,
            ?target,
            ?backfill_parallelism,
            "restoring parallelism after backfill via reschedule"
        );
        let policy = ReschedulePolicy::Parallelism(ParallelismPolicy {
            parallelism: target,
        });
        if let Err(e) = self.reschedule_streaming_job(job_id, policy, false).await {
            tracing::warn!(job_id = %job_id, error = %e.as_report(), "reschedule after backfill failed");
            return Err(e);
        }

        tracing::info!(job_id = %job_id, "parallelism reschedule after backfill submitted");

Review comment (suggestion):
The new private method `apply_post_backfill_parallelism` is missing documentation. Although it's private, it implements important business logic for restoring parallelism after backfill completion. Please add a doc comment explaining its purpose, parameters, return value, and the conditions under which it performs the reschedule.

Code after:
                    job_id = %job_id,
                    ?target,
                    "no backfill parallelism configured, skip post-backfill reschedule"
                );
                return Ok(());
            }
        }

        // Reschedule the job to restore its target parallelism.
        tracing::info!(
            job_id = %job_id,
            ?target,
            ?backfill_parallelism,
            "restoring parallelism after backfill via reschedule"
        );
        let policy = ReschedulePolicy::Parallelism(ParallelismPolicy {
            parallelism: target,
        });
        if let Err(e) = self.reschedule_streaming_job(job_id, policy, false).await {
            tracing::warn!(job_id = %job_id, error = %e.as_report(), "reschedule after backfill failed");
            return Err(e);
        }

        tracing::info!(job_id = %job_id, "parallelism reschedule after backfill submitted");
        Ok(())
    }

    pub fn start_auto_parallelism_monitor(
        self: Arc<Self>,
    ) -> (JoinHandle<()>, oneshot::Sender<()>) {
        tracing::info!("Automatic parallelism scale-out is enabled for streaming jobs");
        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
        let join_handle = tokio::spawn(async move {
            self.run(shutdown_rx).await;
        });

        (join_handle, shutdown_tx)
    }
}