{"record":{"id":"0c065be8060e8f84","repo":"risingwavelabs/risingwave","slug":"receive-commit-request-on-epoch-from-handle","errorCode":null,"errorMessage":"receive commit request on epoch {} from handle {} during alter parallelism","messagePattern":"receive commit request 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":484,"sourceCode":"            .filter(|handle_id| !requests.handle_ids.contains(handle_id))\n            .cloned()\n            .collect();\n        while !remaining_handles.is_empty() || !requests.aligned() {\n            let (handle_id, event) = self.next_event().await?;\n            match event {\n                CoordinationHandleManagerEvent::NewHandle => {\n                    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","sourceCodeStart":466,"sourceCodeEnd":502,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L466-L502","documentation":"During `alter_parallelisms` the coordinator is renegotiating parallelism and expects only Stop/NewHandle/Stop events from handles. Receiving a `CommitRequest` means a sink writer tried to commit data on some epoch mid-renegotiation, which would corrupt the parallelism-change protocol, so the coordinator bails out with this error.","triggerScenarios":"A sink writer that has not yet been stopped (and not yet restarted with the new parallelism) sends a commit request while `alter_parallelisms` is in progress — e.g. the writer did not receive/process the Stop event, or a new handle became active before the coordinator finished the alignment phase.","commonSituations":"Sink writers lagging behind coordinator control messages (slow network, backpressure); a writer restarted with stale state that resumes committing immediately; parallelism changes performed while heavy commit traffic is in flight.","solutions":["Confirm the writer honors the Stop request before it can emit further commits; check the writer's event loop for buffered commits sent after Stop was queued.","Retry the ALTER (e.g. alter sink parallelism) after the sink has quiesced; the coordinator aborts safely on this error.","Investigate ordering between control channel and commit channel on the writer side — commits must be gated behind the parallelism-alignment ack.","Check logs for which handle_id committed and whether it is a leftover handle that should have been stopped first."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// before altering parallelism, verify the sink is quiesced\n// (no pending commits) via the coordinator metrics/logs","typeGuard":"fn is_during_alter(phase: &Phase) -> bool { matches!(phase, Phase::AlteringParallelisms) }","tryCatchPattern":"// retry the alter statement after the error; coordinator state is safely aborted\nif err.contains(\"during alter parallelism\") { sleep(backoff); retry_alter(); }","preventionTips":["Stop all sink writers before changing parallelism.","Gate commit sends behind the parallelism-alignment acknowledgement on the writer side.","Avoid altering parallelism while the sink is under heavy commit backpressure."],"tags":["rust","sink-coordination","parallelism","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"}