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

Code before:
    }

    fn failed_committed(&mut self, epoch: u64, err: SinkError) {
        assert_eq!(self.prepared_epochs.front().expect("non-empty").0, epoch,);
        if let Some((prev_fut, strategy)) = &mut self.backoff_state {
            let new_fut = strategy.next().expect("infinite");
            *prev_fut = new_fut;
        } else {
            let mut strategy = Self::get_retry_backoff_strategy();
            let backoff_fut = strategy.next().expect("infinite");
            self.backoff_state = Some((backoff_fut, strategy));
        }
        tracing::error!(
            error = %err.as_report(),
            %self.sink_id,
            "failed to commit epoch {}, Retrying after backoff",
            epoch,
        );
    }

    async fn try_commit(
        &mut self,
        coordinator: &mut BoxTwoPhaseCoordinator,
        epoch: u64,
        metadata: Vec<u8>,
    ) -> anyhow::Result<()> {
        let start_time = Instant::now();
        let commit_res = run_future_with_periodic_fn(
            coordinator.commit(epoch, metadata),
            Duration::from_secs(5),
            || {
                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(())
    }

Review comment (bug):
Consider adding a doc comment to clarify this method's behavior:
```rust
/// Attempts to commit an epoch to the external sink.
///
/// On success, removes the epoch from `prepared_epochs` and persists the commit.
/// On failure, sets up exponential backoff retry state but returns Ok(()).
/// The retry will occur when `next_to_commit()` is called again.
///
/// Always returns Ok(()) - errors are handled through the retry mechanism.
async fn try_commit(
```

Code after:
    }

    fn failed_committed(&mut self, epoch: u64, err: SinkError) {
        assert_eq!(self.prepared_epochs.front().expect("non-empty").0, epoch,);
        if let Some((prev_fut, strategy)) = &mut self.backoff_state {
            let new_fut = strategy.next().expect("infinite");
            *prev_fut = new_fut;
        } else {
            let mut strategy = Self::get_retry_backoff_strategy();
            let backoff_fut = strategy.next().expect("infinite");
            self.backoff_state = Some((backoff_fut, strategy));
        }
        tracing::error!(
            error = %err.as_report(),
            %self.sink_id,
            "failed to commit epoch {}, Retrying after backoff",
            epoch,
        );
    }

    fn is_empty(&self) -> bool {
        self.pending_epochs.is_empty() && self.prepared_epochs.is_empty()
    }
}

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!(
                    "fail to start {:?} for handle {}",
                    log_store_rewind_start_epoch,
                    handle_id
                )
            })?;
        }