{"record":{"id":"bb28feca5841a6a0","repo":"risingwavelabs/risingwave","slug":"get-none-metadata-in-commit-response-for-coordinat","errorCode":null,"errorMessage":"get none metadata in commit response for coordinated sink writer","messagePattern":"get none metadata in commit response for coordinated sink writer","errorType":"exception","errorClass":"SinkError::Remote","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/remote.rs","lineNumber":662,"sourceCode":"        self.batch_id += 1;\n        Ok(())\n    }\n\n    async fn begin_epoch(&mut self, epoch: u64) -> Result<()> {\n        self.epoch = Some(epoch);\n        Ok(())\n    }\n\n    async fn barrier(&mut self, is_checkpoint: bool) -> Result<Option<SinkMetadata>> {\n        let epoch = self.epoch.ok_or_else(|| {\n            SinkError::Remote(anyhow!(\"epoch has not been initialize, call `begin_epoch`\"))\n        })?;\n        if is_checkpoint {\n            // TODO: add metrics to measure commit time\n            let rsp = self.stream_handle.commit(epoch).await?;\n            rsp.metadata\n                .ok_or_else(|| {\n                    SinkError::Remote(anyhow!(\n                        \"get none metadata in commit response for coordinated sink writer\"\n                    ))\n                })\n                .map(Some)\n        } else {\n            self.stream_handle.barrier(epoch).await?;\n            Ok(None)\n        }\n    }\n}\n\npub struct RemoteCoordinator {\n    stream_handle: SinkCoordinatorStreamHandle,\n}\n\nimpl RemoteCoordinator {\n    pub async fn new<R: RemoteSinkTrait>(param: SinkParam) -> Result<Self> {\n        let stream_handle = EmbeddedConnectorClient::new()?","sourceCodeStart":644,"sourceCodeEnd":680,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/remote.rs#L644-L680","documentation":"During a checkpoint barrier, `RemoteSinkWriter::barrier` calls `stream_handle.commit(epoch)` on the coordinated sink writer service and expects the commit response to carry sink metadata. If the response has `metadata: None`, this error is thrown because the coordinator contract requires metadata on commit.","triggerScenarios":"A checkpoint commit against a coordinated (distributed) sink writer whose `SinkWriterStreamResponse::Commit` reply contains no metadata — e.g. the remote writer implementation returned an empty commit response.","commonSituations":"Custom or buggy sink implementations behind the coordinated writer service; version mismatches between frontend/stream and the sink service; serialization dropping the metadata field.","solutions":["Inspect the remote sink writer service implementation to ensure it always attaches metadata to Commit responses","Verify client and service versions match the expected SinkWriterStreamResponse protocol","Log the raw commit response to identify which writer/service returns None metadata"],"exampleFix":"// before (service side)\nOk(SinkWriterStreamResponse { response: Some(Response::Commit(SinkWriterCommitResponse { metadata: None })) })\n// after\nOk(SinkWriterStreamResponse { response: Some(Response::Commit(SinkWriterCommitResponse { metadata: Some(metadata) })) })","handlingStrategy":"try-catch","validationCode":"// before committing, verify the service implementation returns metadata\nassert!(commit_response.metadata.is_some(), \"coordinated sink must return commit metadata\");","typeGuard":"fn has_metadata(rsp: &SinkWriterStreamResponse) -> bool {\n    matches!(&rsp.response, Some(Response::Commit(c)) if c.metadata.is_some())\n}","tryCatchPattern":"match writer.barrier(true).await {\n    Ok(_) => {},\n    Err(e) if e.to_string().contains(\"get none metadata\") => log::error!(\"sink service returned empty commit metadata: {}\", e),\n    Err(e) => return Err(e),\n}","preventionTips":["Keep client and sink service protobuf definitions in lockstep","Write contract tests asserting Commit responses always carry metadata","Version-gate coordinated sink deployments"],"tags":["rust","risingwave","sink","grpc","protocol"],"backgroundTag":"unexpected-api-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"}