{"record":{"id":"1f05f79d9bd42c27","repo":"risingwavelabs/risingwave","slug":"receive-aligninitialepoch-on-epoch-from-handle","errorCode":null,"errorMessage":"receive AlignInitialEpoch on epoch {} from handle {} during alter parallelism","messagePattern":"receive AlignInitialEpoch on epoch (.+?) from handle (.+?) during alter parallelism","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":491,"sourceCode":"                    requests.add_new_request(handle_id, (), self.vnode_bitmap(handle_id))?;\n                }\n                CoordinationHandleManagerEvent::UpdateVnodeBitmap => {\n                    assert!(remaining_handles.remove(&handle_id));\n                    requests.add_new_request(handle_id, (), self.vnode_bitmap(handle_id))?;\n                }\n                CoordinationHandleManagerEvent::Stop => {\n                    assert!(remaining_handles.remove(&handle_id));\n                    self.stop_handle(handle_id)?;\n                }\n                CoordinationHandleManagerEvent::CommitRequest { epoch, .. } => {\n                    bail!(\n                        \"receive commit request on epoch {} from handle {} during alter parallelism\",\n                        epoch,\n                        handle_id\n                    );\n                }\n                CoordinationHandleManagerEvent::AlignInitialEpoch(epoch) => {\n                    bail!(\n                        \"receive AlignInitialEpoch on epoch {} from handle {} during alter parallelism\",\n                        epoch,\n                        handle_id\n                    );\n                }\n            }\n        }\n        Ok(requests.handle_ids)\n    }\n}\n\n/// Represents the coordinator worker's state machine for handling schema changes.\n///\n/// - `Running`: Normal operation, handles can be started immediately\n/// - `WaitingForFlushed`: Waiting for all pending two-phase commits to complete before starting new handles. This\n///   ensures new sink executors load the correct schema.\nenum CoordinatorWorkerState {\n    Running,","sourceCodeStart":473,"sourceCodeEnd":509,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L473-L509","documentation":"During `alter_parallelisms`, receiving an `AlignInitialEpoch` event is unexpected: epoch alignment is an initialization-time event, but the coordinator is in the parallelism-change phase where existing handles only re-register via NewHandle. The coordinator bails to protect the protocol.","triggerScenarios":"A handle sends `AlignInitialEpoch` while parallelism alteration is running — typically a brand-new handle that skipped the init handshake (e.g. registered after init completed) and thinks it must align epochs, or a writer reusing an old session's alignment flow.","commonSituations":"Sink worker restarts overlapping with a parallelism change; new writers spawned mid-alter that bypass `wait_init_handles`; version-skew between a writer binary that still performs initial alignment and a coordinator that no longer expects it in this phase.","solutions":["Route new handles spawned during `alter_parallelisms` through the correct NewHandle path so they do not run the initial-alignment flow.","Make the writer send AlignInitialEpoch only when it receives an AlignInitialEpochResponse request from the coordinator, never proactively.","Ensure writer and coordinator versions match (rolling-upgrade skew) — check for recent protocol changes.","Retry the alter operation once no writers are (re)connecting."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// writer-side precheck\nif !self.init_phase {\n    // never send AlignInitialEpoch outside init\n    return Err(anyhow!(\"alignment only valid during init\"));\n}","typeGuard":"fn expects_alignment(phase: &Phase) -> bool { matches!(phase, Phase::Init | Phase::WaitingInitHandles) }","tryCatchPattern":"// retry the alter after all writers have re-registered through the proper path\nmatch alter_result { Err(e) if e.contains(\"AlignInitialEpoch\") => retry_after_restart(), _ => {} }","preventionTips":["Spawn new writers during an alter only via the NewHandle path.","Writers should send AlignInitialEpoch only in response to a coordinator request.","Keep writer and coordinator versions in sync during rolling upgrades."],"tags":["rust","sink-coordination","protocol-violation","epoch-alignment"],"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-14T16:17:12.679Z"}