Repository: risingwavelabs/risingwave
File: src/meta/src/manager/sink_coordination/coordinator_worker.rs (Rust)

Code before:
                        )
                        .await?;
                        self.handle_manager
                            .ack_commit(epoch, commit_requests.handle_ids)?;

                        let two_phase_handler = two_phase_handler.as_mut().expect("should exist");
                        two_phase_handler.push_new_item(epoch, commit_metadata);
                    }
                }
            }
        }
    }

    /// Return the log store rewind start epoch if exists.
    async fn init_state_from_store(
        &mut self,
        db: &DatabaseConnection,
        sink_id: SinkId,
        subscriber: SinkCommittedEpochSubscriber,
        coordinator: &mut BoxTwoPhaseCoordinator,
    ) -> anyhow::Result<TwoPhaseCommitHandler> {
        let ordered_metadata = list_sink_states_ordered_by_epoch(db, sink_id as _).await?;

        let mut metadata_iter = ordered_metadata.into_iter().peekable();
        let last_committed_epoch = metadata_iter
            .next_if(|(_, state, _)| matches!(state, SinkState::Committed))
            .map(|(epoch, _, _)| epoch);
        self.prev_committed_epoch = last_committed_epoch;

        let pending_items = metadata_iter
            .peeking_take_while(|(_, state, _)| matches!(state, SinkState::Pending))
            .map(|(epoch, _, metadata)| (epoch, metadata))
            .collect_vec();

        let mut aborted_epochs = vec![];

        for (epoch, state, metadata) in metadata_iter {
            match state {
                SinkState::Aborted => {
                    coordinator.abort(epoch, metadata).await;
                    aborted_epochs.push(epoch);
                }
                other => {
                    unreachable!(
                        "unexpected state {:?} after pending items at epoch {}",
                        other, epoch
                    );
                }
            }
        }


Review comment (suggestion):
This doc comment (line 817 in the full file) is outdated. The method signature changed to only return `TwoPhaseCommitHandler` instead of `(Option<u64>, TwoPhaseCommitHandler)`. The log store rewind start epoch is now stored in `self.prev_committed_epoch` at line 831. Consider updating the doc comment to reflect this change.

Code after:
                        persist_pre_commit_metadata(
                            &db,
                            sink_id as _,
                            epoch,
                            commit_metadata.clone(),
                        )
                        .await?;
                        self.handle_manager
                            .ack_commit(epoch, commit_requests.handle_ids)?;

                        let two_phase_handler = two_phase_handler.as_mut().expect("should exist");
                        two_phase_handler.push_new_item(epoch, commit_metadata);
                    }
                }
            }
        }
    }

    /// Return `TwoPhaseCommitHandler` initialized from the persisted state in the meta store.
    async fn init_state_from_store(
        &mut self,
        db: &DatabaseConnection,
        sink_id: SinkId,
        subscriber: SinkCommittedEpochSubscriber,
        coordinator: &mut BoxTwoPhaseCoordinator,
    ) -> anyhow::Result<TwoPhaseCommitHandler> {
        let ordered_metadata = list_sink_states_ordered_by_epoch(db, sink_id as _).await?;

        let mut metadata_iter = ordered_metadata.into_iter().peekable();
        let last_committed_epoch = metadata_iter
            .next_if(|(_, state, _)| matches!(state, SinkState::Committed))
            .map(|(epoch, _, _)| epoch);
        self.prev_committed_epoch = last_committed_epoch;

        let pending_items = metadata_iter
            .peeking_take_while(|(_, state, _)| matches!(state, SinkState::Pending))
            .map(|(epoch, _, metadata)| (epoch, metadata))
            .collect_vec();

        let mut aborted_epochs = vec![];

        for (epoch, state, metadata) in metadata_iter {
            match state {
                SinkState::Aborted => {
                    coordinator.abort(epoch, metadata).await;
                    aborted_epochs.push(epoch);
                }
                other => {
                    unreachable!(
                        "unexpected state {:?} after pending items at epoch {}",
                        other, epoch