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

Code before:
                    %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)
                .ok_or_else(|| anyhow!("fail to find handle for {} to start", handle_id,))?;
            handle.start(log_store_rewind_start_epoch).map_err(|_| {
                anyhow!(

Review comment (bug):
The `flush_all_pending_items` method will retry commits indefinitely with exponential backoff if the external sink is unavailable. While this may be intentional (based on the PR description stating "wait for all records to commit"), it could cause the system to hang indefinitely during critical operations like alter parallelism (lines 703, 724) and initialization (line 881).

Consider adding:
1. A maximum retry limit or overall timeout to prevent indefinite blocking
2. Documentation explaining the blocking behavior and when it's acceptable
3. A mechanism to cancel the flush operation if the coordinator worker is being shut down

The concern is that `try_commit` always returns `Ok(())` (line 269) even when commit fails, causing the loop to continue indefinitely with the same failing epoch being retried via `next_to_commit` (line 277).
```suggestion
    /// Try to commit the given epoch, retrying up to `max_retries` times on failure.
    /// Returns an error if the commit fails after all retries.
    async fn try_commit(
        &mut self,
        coordinator: &mut BoxTwoPhaseCoordinator,
        epoch: u64,
        metadata: Vec<u8>,
        max_retries: usize,
    ) -> anyhow::Result<()> {
        let start_time = Instant::now();
        let mut attempt = 0;
        loop {
            let commit_res = run_future_with_periodic_fn(
                coordinator.commit(epoch, metadata.clone()),
                Duration::from_secs(5),
                || {
                    warn!(
                        elapsed = ?start_time.elapsed(),
                        %self.sink_id,
                        attempt,
                        "committing during try_commit"
                    );
                },
            )
            .await;

            match commit_res {
                Ok(_) => {
                    self.ack_committed(epoch).await?;
                    return Ok(());
                }
                Err(e) => {
                    self.failed_committed(epoch, e);
                    attempt += 1;
                    if attempt >= max_retries {
                        return Err(anyhow!(
                            "Failed to commit epoch {} after {} attempts for sink {}",
                            epoch,
                            max_retries,
                            self.sink_id
                        ));
                    }
                }
            }
        }
    }

    /// Flush all pending items, retrying each commit up to MAX_COMMIT_RETRIES times.
    /// Returns an error if any commit fails after all retries.
    async fn flush_all_pending_items(
        &mut self,
        coordinator: &mut BoxTwoPhaseCoordinator,
    ) -> anyhow::Result<()> {
        // Maximum number of commit retries per epoch before giving up.
        const MAX_COMMIT_RETRIES: usize = 10;
        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, MAX_COMMIT_RETRIES).await?;
```

Code after:
    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
                )
            })?;
            handle.ack_commit(epoch).map_err(|_| {
                anyhow!(