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

Code before:
            tracker.apply_progress(progress, version_stats);
        }
        for progress in resps
            .iter()
            .flat_map(|resp| &resp.cdc_table_backfill_progress)
        {
            let Some(job_id) = self.fragment_location.get(&progress.fragment_id) else {
                warn!(
                    "update the cdc progress of an non-existent creating streaming job: {progress:?}, which could be cancelled"
                );
                continue;
            };
            let Some(tracker) = &mut self
                .jobs
                .get_mut(job_id)
                .expect("should exist")
                .cdc_table_backfill_tracker
            else {
                warn!("update the progress of an created streaming job: {progress:?}");
                continue;
            };
            tracker.update_split_progress(progress);
        }
        // Handle CDC source offset updated events
        for cdc_offset_updated in resps
            .iter()
            .flat_map(|resp| &resp.cdc_source_offset_updated)
        {
            use risingwave_common::id::SourceId;
            let source_id = SourceId::new(cdc_offset_updated.source_id);
            let job_id = source_id.as_share_source_job_id();
            if let Some(job) = self.jobs.get_mut(&job_id) {
                if let CreateStreamingJobStatus::Creating { tracker, .. } = &mut job.status {
                    tracker.mark_cdc_source_finished();
                }
            } else {
                warn!(
                    "update cdc source offset for non-existent creating streaming job: source_id={}, job_id={}",
                    cdc_offset_updated.source_id, job_id
                );
            }
        }
    }

    fn iter_creating_job_tracker(&self) -> impl Iterator<Item = &CreateMviewProgressTracker> {
        self.jobs.values().filter_map(|job| match &job.status {
            CreateStreamingJobStatus::Init => None,
            CreateStreamingJobStatus::Creating { tracker, .. } => Some(tracker),
            CreateStreamingJobStatus::Created => None,
        })
    }

Review comment (bug):
The CDC source offset update handler should also handle the case where the job exists but is not in the Creating state, similar to how cdc_table_backfill_progress is handled above. Consider adding a check for the job status and logging a warning if it's not in the Creating state, to help with debugging if offset updates arrive for jobs that have already been created or are in the Init state.
```suggestion
                    tracker.mark_cdc_source_finished();
                } else {
                    warn!(
                        "update cdc source offset for streaming job not in Creating state: source_id={}, job_id={}, status={:?}",
                        cdc_offset_updated.source_id, job_id, job.status
                    );
```

Code after:
                    "update the cdc progress of an non-existent creating streaming job: {progress:?}, which could be cancelled"
                );
                continue;
            };
            let Some(tracker) = &mut self
                .jobs
                .get_mut(job_id)
                .expect("should exist")
                .cdc_table_backfill_tracker
            else {
                warn!("update the progress of an created streaming job: {progress:?}");
                continue;
            };
            tracker.update_split_progress(progress);
        }
        // Handle CDC source offset updated events
        for cdc_offset_updated in resps
            .iter()
            .flat_map(|resp| &resp.cdc_source_offset_updated)
        {
            use risingwave_common::id::SourceId;
            let source_id = SourceId::new(cdc_offset_updated.source_id);
            let job_id = source_id.as_share_source_job_id();
            if let Some(job) = self.jobs.get_mut(&job_id) {
                if let CreateStreamingJobStatus::Creating { tracker, .. } = &mut job.status {
                    tracker.mark_cdc_source_finished();
                }
            } else {
                warn!(
                    "update cdc source offset for non-existent creating streaming job: source_id={}, job_id={}",
                    cdc_offset_updated.source_id, job_id
                );
            }
        }
    }

    fn iter_creating_job_tracker(&self) -> impl Iterator<Item = &CreateMviewProgressTracker> {
        self.jobs.values().filter_map(|job| match &job.status {
            CreateStreamingJobStatus::Init => None,
            CreateStreamingJobStatus::Creating { tracker, .. } => Some(tracker),
            CreateStreamingJobStatus::Created => None,
        })
    }

    fn iter_mut_creating_job_tracker(
        &mut self,
    ) -> impl Iterator<Item = &mut CreateMviewProgressTracker> {
        self.jobs
            .values_mut()
            .filter_map(|job| match &mut job.status {
                CreateStreamingJobStatus::Init => None,