{"record":{"id":"e447ac6b0f63a24a","repo":"risingwavelabs/risingwave","slug":"expect-new-handle-during-init-but-got","errorCode":null,"errorMessage":"expect new handle during init, but got {}","messagePattern":"expect new handle during init, but got (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":447,"sourceCode":"        self.writer_handles\n            .remove(&handle_id)\n            .expect(\"should exist\")\n            .stop()\n    }\n\n    async fn wait_init_handles(&mut self) -> anyhow::Result<HashSet<HandleId>> {\n        assert!(self.writer_handles.is_empty());\n        let mut init_requests = AligningRequests::default();\n        while !init_requests.aligned() {\n            let (handle_id, event) = self.next_event().await?;\n            let unexpected_event = match event {\n                CoordinationHandleManagerEvent::NewHandle => {\n                    init_requests.add_new_request(handle_id, (), self.vnode_bitmap(handle_id))?;\n                    continue;\n                }\n                event => event.name(),\n            };\n            return Err(anyhow!(\n                \"expect new handle during init, but got {}\",\n                unexpected_event\n            ));\n        }\n        Ok(init_requests.handle_ids)\n    }\n\n    async fn alter_parallelisms(\n        &mut self,\n        altered_handles: impl Iterator<Item = HandleId>,\n    ) -> anyhow::Result<HashSet<HandleId>> {\n        let mut requests = AligningRequests::default();\n        for handle_id in altered_handles {\n            requests.add_new_request(handle_id, (), self.vnode_bitmap(handle_id))?;\n        }\n        let mut remaining_handles: HashSet<_> = self\n            .writer_handles\n            .keys()","sourceCodeStart":429,"sourceCodeEnd":465,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L429-L465","documentation":"`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.","triggerScenarios":"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`.","commonSituations":"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.","solutions":["Inspect the reported event name in the message to see which protocol stage fired too early; that tells which writer misbehaved.","Fix the writer side so it waits for the StartCoordinationResponse (and AlignInitialEpoch ack when enabled) before sending CommitRequest or AlignInitialEpoch.","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.","Check for races where a previous coordinator session's writers reconnect and emit stale events."],"exampleFix":"// before\nreturn Err(anyhow!(\n    \"expect new handle during init, but got {}\",\n    unexpected_event\n));\n// after\nCoordinationHandleManagerEvent::Stop => {\n    tracing::warn!(?handle_id, \"writer stopped during init; dropping it\");\n    continue;\n}","handlingStrategy":"try-catch","validationCode":"// writer-side precheck: only send NewHandle until StartCoordinationResponse is received\nassert!(self.awaiting_start, \"writer not in init phase\");","typeGuard":"fn is_new_handle(ev: &CoordinationHandleManagerEvent) -> bool { matches!(ev, CoordinationHandleManagerEvent::NewHandle) }","tryCatchPattern":"// coordinator\nmatch res {\n    Err(e) if e.to_string().contains(\"expect new handle during init\") => {\n        // inspect offending event, restart affected writer\n    }\n    other => other?,\n}","preventionTips":["Keep the writer state machine strictly ordered: NewHandle registration -> StartResponse -> optional AlignInitialEpoch -> commits.","Never reuse writer sessions from a previous coordinator incarnation.","Log every event the writer sends with its phase label."],"tags":["rust","protocol-state-machine","sink-coordination"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}