Repository: risingwavelabs/risingwave
File: src/stream/src/executor/source/source_executor.rs (Rust)

Code before:

        stream.replace_data_stream(reader_stream);

        Ok(())
    }

    async fn persist_state_and_clear_cache(
        &mut self,
        epoch: EpochPair,
    ) -> StreamExecutorResult<HashMap<SplitId, SplitImpl>> {
        let core = &mut self.stream_source_core;

        let cache = core
            .updated_splits_in_epoch
            .values()
            .map(|split_impl| split_impl.to_owned())
            .collect_vec();

        if !cache.is_empty() {
            tracing::debug!(state = ?cache, "take snapshot");

            // Record metrics for CDC sources before moving cache
            let source_id = core.source_id.to_string();
            for split_impl in &cache {
                // PostgreSQL CDC: Extract and record LSN
                if let Some(state_table_lsn_value) = extract_split_offset(split_impl) {
                    self.metrics
                        .pg_cdc_state_table_lsn
                        .with_guarded_label_values(&[&source_id])
                        .set(state_table_lsn_value as i64);
                }

                // MySQL CDC: Extract and record binlog file sequence and position
                if let SplitImpl::MysqlCdc(mysql_split) = split_impl
                    && let Some(offset_str) = mysql_split.start_offset().as_ref()
                    && let Some((file_seq, position)) = extract_mysql_cdc_binlog_offset(offset_str)
                {
                    self.metrics
                        .mysql_cdc_state_binlog_file_seq
                        .with_guarded_label_values(&[&source_id])
                        .set(file_seq as i64);

                    self.metrics
                        .mysql_cdc_state_binlog_position
                        .with_guarded_label_values(&[&source_id])
                        .set(position as i64);
                }
            }

            core.split_state_store.set_states(cache).await?;
        }

Review comment (bug):
Related to #22968: the naming `extract_split_offset` here is confusing. It is specific to PG CDC bug the naming suggests that it is for connector. What we do in this PR by checking the SplitImpl enum first is clearer to me, let's change the PG part as well
```
match split {
        SplitImpl::PostgresCdc(pg_split) => {
            if let Some(pg_lsn) = pg_split.pg_lsn() {
              ...
            }
            
        },
        SplitImpl::MysqlCdc(mysql_split) => {
          if let Some((file_seq, position)) = mysql_split.mysql_binlog_offset() {
             ...
          }
        }
                    
        _ => None,
    }
```

Code after:

        // commit anyway, even if no message saved
        core.split_state_store.commit(epoch).await?;

        let updated_splits = core.updated_splits_in_epoch.clone();

        core.updated_splits_in_epoch.clear();

        Ok(updated_splits)
    }

    /// try mem table spill
    async fn try_flush_data(&mut self) -> StreamExecutorResult<()> {
        let core = &mut self.stream_source_core;
        core.split_state_store.try_flush().await?;

        Ok(())
    }

    /// A source executor with a stream source receives:
    /// 1. Barrier messages
    /// 2. Data from external source
    /// and acts accordingly.
    #[try_stream(ok = Message, error = StreamExecutorError)]
    async fn execute_inner(mut self) {
        let mut barrier_receiver = self.barrier_receiver.take().unwrap();
        let first_barrier = barrier_receiver
            .recv()
            .instrument_await("source_recv_first_barrier")
            .await
            .ok_or_else(|| {
                anyhow!(
                    "failed to receive the first barrier, actor_id: {:?}, source_id: {:?}",
                    self.actor_ctx.id,
                    self.stream_source_core.source_id
                )
            })?;
        let first_epoch = first_barrier.epoch;
        let mut boot_state =
            if let Some(splits) = first_barrier.initial_split_assignment(self.actor_ctx.id) {
                tracing::debug!(?splits, "boot with splits");
                splits.to_vec()
            } else {
                Vec::default()
            };
        let is_pause_on_startup = first_barrier.is_pause_on_startup();
        let mut is_uninitialized = first_barrier.is_newly_added(self.actor_ctx.id);

        yield Message::Barrier(first_barrier);

        let mut core = self.stream_source_core;