{"record":{"id":"5ed222ecb3014df7","repo":"risingwavelabs/risingwave","slug":"receive-aligninitialepoch-after-initialization","errorCode":null,"errorMessage":"receive AlignInitialEpoch after initialization","messagePattern":"receive AlignInitialEpoch after initialization","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":731,"sourceCode":"                    }\n                    CoordinationHandleManagerEvent::Stop => {\n                        self.handle_manager.stop_handle(handle_id)?;\n                        running_handles = self\n                            .handle_manager\n                            .alter_parallelisms(pending_new_handles.drain(..))\n                            .await?;\n                        self.try_handle_init_requests(&running_handles, &mut two_phase_handler)\n                            .await?;\n\n                        continue;\n                    }\n                    CoordinationHandleManagerEvent::CommitRequest {\n                        epoch,\n                        metadata,\n                        schema_change,\n                    } => (handle_id, epoch, (metadata, schema_change)),\n                    CoordinationHandleManagerEvent::AlignInitialEpoch(_) => {\n                        bail!(\"receive AlignInitialEpoch after initialization\")\n                    }\n                },\n                CoordinatorWorkerEvent::ReadyToCommit(epoch, metadata, schema_change) => {\n                    let start_time = Instant::now();\n                    let commit_fut = async {\n                        match &mut coordinator {\n                            SinkCommitCoordinator::SinglePhase(coordinator) => {\n                                assert!(metadata.is_none());\n                                if let Some(schema_change) = schema_change {\n                                    coordinator\n                                        .commit_schema_change(epoch, schema_change)\n                                        .instrument_await(Self::commit_span(\n                                            \"single_phase_schema_change\",\n                                            sink_id,\n                                            epoch,\n                                        ))\n                                        .await?;\n                                }","sourceCodeStart":713,"sourceCodeEnd":749,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L713-L749","documentation":"In the steady-state coordination loop (`run_coordination`), receiving `AlignInitialEpoch` after initialization has completed is invalid — epoch alignment only happens during the init handshake. The coordinator bails because continuing would mean an unaligned writer joins mid-stream, which can break epoch ordering.","triggerScenarios":"A handle that joins after the coordinator finished init sends an AlignInitialEpoch event — e.g. a writer reconnecting and replaying its startup sequence, or a writer that missed the transition from init to running phase.","commonSituations":"Writer reconnect after network blip that replays old handshake logic; coordinator failover where a writer thinks init is still in progress; new sink workers added without going through a fresh coordinator init round.","solutions":["Fix the writer's reconnect logic: after restart it should wait for the coordinator to run a new init/alignment round rather than sending AlignInitialEpoch spontaneously.","Restart the sink so writer and coordinator resynchronize through the normal handshake.","Check coordinator failover logic — after meta failover the coordinator should re-run `wait_init_handles` for existing handles.","Verify rolling upgrades: an older writer binary may still send AlignInitialEpoch in steady state."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// writer-side precheck before sending alignment\nif !self.session.is_init_phase() { return Err(anyhow!(\"alignment only during init\")); }","typeGuard":"fn in_steady_state(phase: &Phase) -> bool { matches!(phase, Phase::Running) }","tryCatchPattern":"// on this error, restart the sink so writer and coordinator redo the handshake\nif err.contains(\"receive AlignInitialEpoch after initialization\") { restart_sink(); }","preventionTips":["Fix writer reconnect logic to wait for a fresh init round instead of replaying startup events.","On coordinator failover, re-run wait_init_handles for existing handles.","Avoid mixed writer/coordinator versions during upgrades."],"tags":["rust","sink-coordination","epoch-alignment","protocol-violation"],"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"}