risingwavelabs/risingwave · error

expect new handle during init, but got

Error message

expect new handle during init, but got {}

What it means

`wait_init_handles` expects every event received during the initialization phase to be `CoordinationHandleManagerEvent::NewHandle`. Any other event (Stop, CommitRequest, AlignInitialEpoch, etc.) breaks the expected init protocol and yields this error. It signals that sink writers are misbehaving during the initial handshake.

Solutions

1. Inspect the reported event name in the message to see which protocol stage fired too early; that tells which writer misbehaved.
2. Fix the writer side so it waits for the StartCoordinationResponse (and AlignInitialEpoch ack when enabled) before sending CommitRequest or AlignInitialEpoch.
3. Handle writer aborts during init gracefully instead of failing the whole init: treat Stop during init as a normal early termination and remove the handle from init_requests.
4. Check for races where a previous coordinator session's writers reconnect and emit stale events.

Example fix

// before
return Err(anyhow!(
    "expect new handle during init, but got {}",
    unexpected_event
));
// after
CoordinationHandleManagerEvent::Stop => {
    tracing::warn!(?handle_id, "writer stopped during init; dropping it");
    continue;
}

// writer-side precheck: only send NewHandle until StartCoordinationResponse is received
assert!(self.awaiting_start, "writer not in init phase");

fn is_new_handle(ev: &CoordinationHandleManagerEvent) -> bool { matches!(ev, CoordinationHandleManagerEvent::NewHandle) }

// coordinator
match res {
    Err(e) if e.to_string().contains("expect new handle during init") => {
        // inspect offending event, restart affected writer
    }
    other => other?,
}

When it happens

Trigger: A sink writer handle sends a `Stop`, `CommitRequest`, or `AlignInitialEpoch` event before the init handshake completes; i.e. a writer aborts, tries to commit, or re-aligns while the coordinator is still collecting new handles in `wait_init_handles`.

Common situations: A sink writer crashes or its upstream actor is cancelled during initialization and sends Stop; a writer that already aligned in a previous incarnation sends AlignInitialEpoch immediately after reconnect; races between coordinator startup and writer shutdown.

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

• expect AlignInitialEpoch but got
• empty sink metadata
• end of writer request stream
• failed to ack aligned initial epoch
• failed to acknowledge the commit for epoch

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

Appendix: source

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

        self.writer_handles
            .remove(&handle_id)
            .expect("should exist")
            .stop()
    }

    async fn wait_init_handles(&mut self) -> anyhow::Result<HashSet<HandleId>> {
        assert!(self.writer_handles.is_empty());
        let mut init_requests = AligningRequests::default();
        while !init_requests.aligned() {
            let (handle_id, event) = self.next_event().await?;
            let unexpected_event = match event {
                CoordinationHandleManagerEvent::NewHandle => {
                    init_requests.add_new_request(handle_id, (), self.vnode_bitmap(handle_id))?;
                    continue;
                }
                event => event.name(),
            };
            return Err(anyhow!(
                "expect new handle during init, but got {}",
                unexpected_event
            ));
        }
        Ok(init_requests.handle_ids)
    }

    async fn alter_parallelisms(
        &mut self,
        altered_handles: impl Iterator<Item = HandleId>,
    ) -> anyhow::Result<HashSet<HandleId>> {
        let mut requests = AligningRequests::default();
        for handle_id in altered_handles {
            requests.add_new_request(handle_id, (), self.vnode_bitmap(handle_id))?;
        }
        let mut remaining_handles: HashSet<_> = self
            .writer_handles
            .keys()

View on GitHub (pinned to 6469eb736d)