What causes "failed to ack aligned initial epoch {:?} for handle {}" in risingwavelabs/risingwave?

Capture the inner error (replace `map_err(|_| ...)`) to identify whether the writer channel is closed or the state is wrong.

risingwavelabs/risingwave · error

failed to ack aligned initial epoch {:?} for handle {}

Error message

failed to ack aligned initial epoch {:?} for handle {}

What it means

In `ack_aligned_initial_epoch`, every registered writer handle must acknowledge the aligned initial epoch; if any handle's `ack_aligned_initial_epoch` returns `Err`, this wrapper error is raised. It indicates the handle's writer side cannot accept the aligned epoch — usually because the writer actor has terminated or is in a state that rejects epoch alignment (e.g. already started or stopped).

Source

Thrown at src/meta/src/manager/sink_coordination/coordinator_worker.rs:310

                .get_mut(&handle_id)
                .ok_or_else(|| anyhow!("failed to find handle {} to start", handle_id,))?;
            handle.start(log_store_rewind_start_epoch).map_err(|_| {
                anyhow!(
                    "failed 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!(
                        "failed 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!(
                    "failed to find handle {} when acknowledging the commit for epoch {}",

View on GitHub (pinned to 6469eb736d)

Solutions

Capture the inner error (replace `map_err(|_| ...)`) to identify whether the writer channel is closed or the state is wrong.
Confirm the writer task for each handle is running before aligning epochs.
Re-run the initialization flow (`wait_init_handles` then align) if handles were recreated.
Guard against concurrent stop/alter_parallelisms while alignment is in progress.

Example fix

// before
handle.ack_aligned_initial_epoch(epoch).map_err(|_| anyhow!("failed to ack aligned initial epoch ..."))?;

// after
handle.ack_aligned_initial_epoch(epoch)
    .with_context(|| format!("failed to ack aligned initial epoch {} for handle {:?}", epoch, handle_id))?;

Defensive patterns

Strategy: try-catch

Validate before calling

// Ensure all handles are alive before acking alignment
for id in manager.registered_handle_ids() {
    anyhow::ensure!(manager.is_handle_alive(&id), "handle {} not alive", id);
}

Try / catch

match manager.ack_aligned_initial_epoch(epoch) {
    Ok(()) => {},
    Err(e) if e.to_string().contains("failed to ack aligned initial epoch") => {
        warn!(error = ?e, "re-running init flow for sink alignment");
    }
    Err(e) => return Err(e),
}

Prevention

Verify writer tasks are running before epoch alignment
Suppress stop/parallelism-change while alignment is in flight
Capture the inner handle error for diagnosis

When it happens

Trigger: Calling `ack_aligned_initial_epoch` on a handle whose writer was dropped/closed (send to writer fails), or whose state machine is not in the pre-start phase that accepts an aligned initial epoch.

Common situations: Writer task crashed earlier during sink recovery; acknowledgment sent after handles were already stopped by a parallelism change; race between failover and sink re-initialization.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/a1a3fb8b9ccd4e7b. Report an issue: GitHub.