{"record":{"id":"6ddbd61436708e3b","repo":"risingwavelabs/risingwave","slug":"should-get-aligninitialepochresponse-but-get","errorCode":null,"errorMessage":"should get AlignInitialEpochResponse but get {:?}","messagePattern":"should get AlignInitialEpochResponse but get (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/rpc_client/src/sink_coordinate_client.rs","lineNumber":163,"sourceCode":"            CoordinateResponse {\n                msg: Some(coordinate_response::Msg::Stopped(_)),\n            } => Ok(()),\n            msg => Err(anyhow!(\"should get Stopped but get {:?}\", msg)),\n        }\n    }\n\n    pub async fn align_initial_epoch(&mut self, initial_epoch: u64) -> anyhow::Result<u64> {\n        self.send_request(CoordinateRequest {\n            msg: Some(coordinate_request::Msg::AlignInitialEpochRequest(\n                initial_epoch,\n            )),\n        })\n        .await?;\n        match self.next_response().await? {\n            CoordinateResponse {\n                msg: Some(coordinate_response::Msg::AlignInitialEpochResponse(epoch)),\n            } => Ok(epoch),\n            msg => Err(anyhow!(\n                \"should get AlignInitialEpochResponse but get {:?}\",\n                msg\n            )),\n        }\n    }\n}\n","sourceCodeStart":145,"sourceCodeEnd":170,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/rpc_client/src/sink_coordinate_client.rs#L145-L170","documentation":"SinkCoordinateClient::align_initial_epoch sends an AlignInitialEpochRequest and expects an AlignInitialEpochResponse carrying the aligned epoch. Any other response triggers this error, meaning epoch alignment with the sink coordinator did not complete as expected.","triggerScenarios":"Calling align_initial_epoch during sink initialization when the coordinator returns a different CoordinateResponse variant — stale buffered response, coordinator restart, or protocol version skew.","commonSituations":"Sink startup racing with coordinator restart; stream desync from earlier failed calls; compute node/meta version mismatch after upgrade.","solutions":["Read the debug-printed `msg` to see the unexpected response","Recreate the coordination stream and redo initialization (align_initial_epoch) from scratch","Check coordinator logs and align component versions"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Rust\nlet epoch = loop {\n    match client.align_initial_epoch(initial_epoch).await {\n        Ok(e) => break e,\n        Err(e) if e.to_string().contains(\"AlignInitialEpochResponse\") && retries < 3 => {\n            retries += 1;\n            client = SinkCoordinateClient::start(...).await?;\n        }\n        Err(e) => return Err(e),\n    }\n};","preventionTips":["Perform epoch alignment once, at sink init, on a fresh stream","Recreate the stream rather than reusing one after a failed round-trip","Keep compute node and coordinator versions aligned"],"tags":["grpc","protocol","sink","streaming"],"backgroundTag":"unexpected-response-shape","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"}