{"record":{"id":"7c7b4d5ae61c4308","repo":"risingwavelabs/risingwave","slug":"empty-sink-metadata","errorCode":null,"errorMessage":"empty sink metadata","messagePattern":"empty sink metadata","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":402,"sourceCode":"    async fn next_event(&mut self) -> anyhow::Result<(HandleId, CoordinationHandleManagerEvent)> {\n        select! {\n            handle = self.request_rx.recv() => {\n                let handle = handle.ok_or_else(|| anyhow!(\"end of writer request stream\"))?;\n                if handle.param() != &self.param {\n                    warn!(prev_param = ?self.param, new_param = ?handle.param(), \"sink param mismatch\");\n                }\n                let handle_id = self.next_handle_id;\n                self.next_handle_id += 1;\n                self.writer_handles.insert(handle_id, handle);\n                Ok((handle_id, CoordinationHandleManagerEvent::NewHandle))\n            }\n            result = Self::next_request_inner(&mut self.writer_handles) => {\n                let (handle_id, request) = result?;\n                let event = match request {\n                    coordinate_request::Msg::CommitRequest(request) => {\n                        CoordinationHandleManagerEvent::CommitRequest {\n                            epoch: request.epoch,\n                            metadata: request.metadata.ok_or_else(|| anyhow!(\"empty sink metadata\"))?,\n                            schema_change: request.schema_change,\n                        }\n                    }\n                    coordinate_request::Msg::AlignInitialEpochRequest(epoch) => {\n                        CoordinationHandleManagerEvent::AlignInitialEpoch(epoch)\n                    }\n                    coordinate_request::Msg::UpdateVnodeRequest(_) => {\n                        CoordinationHandleManagerEvent::UpdateVnodeBitmap\n                    }\n                    coordinate_request::Msg::Stop(_) => {\n                        CoordinationHandleManagerEvent::Stop\n                    }\n                    coordinate_request::Msg::StartRequest(_) => {\n                        unreachable!(\"should have been handled\");\n                    }\n                };\n                Ok((handle_id, event))\n            }","sourceCodeStart":384,"sourceCodeEnd":420,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L384-L420","documentation":"A `CommitRequest` arrived from a writer handle but its `metadata` field is `None`. Sink metadata (e.g. the external sink's target/status info needed for the commit) is required for every commit, so the manager rejects the request with this error. It indicates the writer sent a malformed or metadata-less commit message over the gRPC coordination stream.","triggerScenarios":"A stream writer builds `coordinate_request::Msg::CommitRequest` without setting `metadata` — e.g. after a protobuf schema change where the field was renamed/newer writers omit it, or a bug in the commit path that forgets to populate `sink_metadata`.","commonSituations":"Version skew between frontend/stream compute and meta node after a protobuf field change; a connector whose sink metadata serialization fails and silently produces `None`; hand-rolled test clients sending incomplete commit requests.","solutions":["Check the writer side (stream executor) to ensure `sink_metadata` is populated before sending `CommitRequest`.","Look for recent protobuf changes to `coordinate_request` and align binaries across the cluster (rolling upgrade skew).","Verify the connector's sink metadata serialization path for the failing sink type (e.g. Kafka/Iceberg).","Add a validation at the writer so a commit without metadata fails fast locally instead of reaching the meta node."],"exampleFix":"// before\nlet msg = coordinate_request::Msg::CommitRequest(CommitRequest {\n    epoch,\n    metadata: None,\n    schema_change: None,\n});\n\n// after\nlet msg = coordinate_request::Msg::CommitRequest(CommitRequest {\n    epoch,\n    metadata: Some(build_sink_metadata(&sink).expect(\"sink metadata required for commit\")),\n    schema_change: None,\n});","handlingStrategy":"validation","validationCode":"// On the writer side, refuse to send commits without metadata\nlet metadata = build_sink_metadata(&sink)\n    .ok_or_else(|| anyhow!(\"sink metadata must be set before commit\"))?;\nlet msg = CommitRequest { epoch, metadata: Some(metadata), schema_change };\nanyhow::ensure!(msg.metadata.is_some(), \"CommitRequest requires metadata\");","typeGuard":"fn has_metadata(req: &CommitRequest) -> bool {\n    req.metadata.is_some()\n}","tryCatchPattern":"if let Err(e) = manager.next_event().await {\n    if e.to_string().contains(\"empty sink metadata\") {\n        error!(\"malformed CommitRequest from writer; check version skew and connector metadata serialization\");\n    }\n    return Err(e);\n}","preventionTips":["Always populate sink_metadata before sending CommitRequest","Keep meta node and stream compute binaries at compatible versions","Validate connector metadata serialization in connector tests"],"tags":["rust","protobuf","missing-field","sink-coordination"],"backgroundTag":"empty-required-field","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"}