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

Code before:
            || {
                warn!(
                    elapsed = ?start_time.elapsed(),
                    %self.sink_id,
                    "committing during try_commit"
                );
            },
        )
        .await;

        match commit_res {
            Ok(_) => {
                self.ack_committed(epoch).await?;
            }
            Err(e) => {
                self.failed_committed(epoch, e);
            }
        }

        Ok(())
    }

    async fn flush_all_pending_items(
        &mut self,
        coordinator: &mut BoxTwoPhaseCoordinator,
    ) -> anyhow::Result<()> {
        while !self.pending_epochs.is_empty() || !self.prepared_epochs.is_empty() {
            let (epoch, metadata) = self.next_to_commit().await?;
            self.try_commit(coordinator, epoch, metadata).await?;
        }
        Ok(())
    }
}

struct CoordinationHandleManager {
    param: SinkParam,
    writer_handles: HashMap<HandleId, SinkWriterCoordinationHandle>,
    next_handle_id: HandleId,
    request_rx: UnboundedReceiver<SinkWriterCoordinationHandle>,
}

impl CoordinationHandleManager {
    fn start(
        &mut self,
        log_store_rewind_start_epoch: Option<u64>,
        handle_ids: impl IntoIterator<Item = HandleId>,
    ) -> anyhow::Result<()> {
        for handle_id in handle_ids {
            let handle = self
                .writer_handles
                .get_mut(&handle_id)

Review comment (bug):
Consider adding a doc comment to explain the behavior of this method, particularly that:
1. It blocks until all pending and prepared epochs are committed
2. It retries failed commits indefinitely with exponential backoff
3. It should only be called during synchronization points (initialization, alter parallelism)

Example:
```rust
/// Flushes all pending items to the external sink, blocking until completion.
///
/// This method processes all items in `pending_epochs` and `prepared_epochs`,
/// committing them to the external sink with retry on failure. It will block
/// indefinitely retrying with exponential backoff if the external sink is unavailable.
///
/// Should only be called during synchronization points such as initialization
/// or alter parallelism operations.
async fn flush_all_pending_items(
```

Code after:
}

impl CoordinationHandleManager {
    fn start(
        &mut self,
        log_store_rewind_start_epoch: Option<u64>,
        handle_ids: impl IntoIterator<Item = HandleId>,
    ) -> anyhow::Result<()> {
        for handle_id in handle_ids {
            let handle = self
                .writer_handles
                .get_mut(&handle_id)
                .ok_or_else(|| anyhow!("fail to find handle for {} to start", handle_id,))?;
            handle.start(log_store_rewind_start_epoch).map_err(|_| {
                anyhow!(
                    "fail to start {:?} for handle {}",
                    log_store_rewind_start_epoch,
                    handle_id
                )
            })?;
        }
        Ok(())
    }

    fn ack_aligned_initial_epoch(&mut self, aligned_initial_epoch: u64) -> anyhow::Result<()> {
        for (handle_id, handle) in &mut self.writer_handles {
            handle
                .ack_aligned_initial_epoch(aligned_initial_epoch)
                .map_err(|_| {
                    anyhow!(
                        "fail to ack_aligned_initial_epoch {:?} for handle {}",
                        aligned_initial_epoch,
                        handle_id
                    )
                })?;
        }
        Ok(())
    }

    fn ack_commit(
        &mut self,
        epoch: u64,
        handle_ids: impl IntoIterator<Item = HandleId>,
    ) -> anyhow::Result<()> {
        for handle_id in handle_ids {
            let handle = self.writer_handles.get_mut(&handle_id).ok_or_else(|| {
                anyhow!(
                    "fail to find handle for {} when ack commit on epoch {}",
                    handle_id,
                    epoch
                )