{"record":{"id":"6ab16b95e3608be7","repo":"risingwavelabs/risingwave","slug":"should-get-stopped-but-get","errorCode":null,"errorMessage":"should get Stopped but get {:?}","messagePattern":"should get Stopped but get (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/rpc_client/src/sink_coordinate_client.rs","lineNumber":148,"sourceCode":"                    Some(coordinate_response::Msg::StartResponse(StartCoordinationResponse {\n                        log_store_rewind_start_epoch,\n                    })),\n            } => Ok(log_store_rewind_start_epoch\n                .ok_or_else(|| anyhow!(\"should get start epoch after update vnode bitmap\"))?),\n            msg => Err(anyhow!(\"should get start response but get {:?}\", msg)),\n        }\n    }\n\n    pub async fn stop(mut self) -> anyhow::Result<()> {\n        self.send_request(CoordinateRequest {\n            msg: Some(coordinate_request::Msg::Stop(true)),\n        })\n        .await?;\n        match self.next_response().await? {\n            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            )),","sourceCodeStart":130,"sourceCodeEnd":166,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/rpc_client/src/sink_coordinate_client.rs#L130-L166","documentation":"SinkCoordinateClient::stop sends a Stop(true) request and expects the coordinator to answer with a Stopped response. Any other response yields this error, meaning the sink coordination session's termination was not confirmed.","triggerScenarios":"Calling stop when the coordinator responds with a non-Stopped message — coordinator already terminated, stream closed with an error-shaped response, or stale response from a prior request.","commonSituations":"Coordinator dropped the stream during shutdown; sink actor tearing down concurrently with a coordinator failure; requests misordered after a failed earlier round.","solutions":["Inspect the debug-printed `msg` to identify the actual response","Treat as best-effort on teardown: if the stream is already closed, the sink can be considered stopped; otherwise recreate the stream and send Stop again","Check coordinator logs for termination errors"],"exampleFix":"// before\nmatch self.next_response().await? { ... Stopped(_) => Ok(()), msg => Err(anyhow!(\"should get Stopped but get {:?}\", msg)) }\n// after (teardown is best-effort)\nmatch self.next_response().await? {\n    CoordinateResponse { msg: Some(coordinate_response::Msg::Stopped(_)) } => Ok(()),\n    other => { tracing::warn!(\"stop: unexpected coordinator response {:?}; treating as stopped\", other); Ok(()) }\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Rust: stop is best-effort during teardown\nif let Err(e) = client.stop().await {\n    tracing::warn!(\"sink stop not confirmed: {e:#}; continuing teardown\");\n}","preventionTips":["Treat stop as idempotent/best-effort — the coordinator may already be gone","Avoid calling stop concurrently with other requests","Check coordinator liveness before structured shutdown"],"tags":["grpc","protocol","sink","teardown"],"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-23T08:17:48.524Z"}