{"record":{"id":"daa58eee05e7ec17","repo":"risingwavelabs/risingwave","slug":"e-daa58e","errorCode":null,"errorMessage":"{e}","messagePattern":"\\{e\\}","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/formatter/mod.rs","lineNumber":178,"sourceCode":"                    )));\n                }\n            }\n        } else {\n            encoder\n        };\n        Ok(encoder)\n    }\n}\n\nimpl EncoderBuild for ProtoEncoder {\n    async fn build(b: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {\n        // TODO: better to be a compile-time assert\n        assert!(pk_indices.is_none());\n        // By passing `None` as `aws_auth_props`, reading from `s3://` not supported yet.\n        let (descriptor, sid) =\n            crate::schema::protobuf::fetch_descriptor(&b.format_desc.options, b.topic, None)\n                .await\n                .map_err(|e| SinkError::Config(anyhow!(e)))?;\n        let header = match sid {\n            None => ProtoHeader::None,\n            Some(sid) => ProtoHeader::ConfluentSchemaRegistry(sid),\n        };\n        ProtoEncoder::new(b.schema, None, descriptor, header)\n    }\n}\n\nfn ensure_only_one_pk<'a>(\n    data_type_name: &'a str,\n    params: &'a EncoderParams<'_>,\n    pk_indices: &'a Option<Vec<usize>>,\n) -> Result<(usize, &'a Field)> {\n    let Some(pk_indices) = pk_indices else {\n        return Err(SinkError::Config(anyhow!(\n            \"{}Encoder requires primary key columns to be specified\",\n            data_type_name\n        )));","sourceCodeStart":160,"sourceCodeEnd":196,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/formatter/mod.rs#L160-L196","documentation":"When building a protobuf encoder, the connector fetches the `.proto` file descriptor (from provided options or the Confluent Schema Registry, using the topic as subject). Any failure in `fetch_descriptor` — unparseable proto content, unreachable registry, missing subject — is wrapped into this Config error with the underlying message.","triggerScenarios":"Building a protobuf-encoded sink where the descriptor cannot be fetched: inline proto option has syntax errors, Schema Registry URL/credentials are wrong, or no schema exists for the topic.","commonSituations":"Typo'd proto definitions, Schema Registry down or auth-protected, subject naming mismatch (`<topic>-value` not registered), and lack of registry connectivity from compute nodes.","solutions":["Check the underlying `{e}` message: fix proto syntax if inline, or registry URL/auth if fetched remotely","Register or correct the schema subject for the topic in Confluent Schema Registry","Validate the inline proto compiles (e.g. with protoc) before using it in the sink","Verify network reachability to the Schema Registry from RisingWave nodes"],"exampleFix":"// before\nWITH (... format = 'plain', encode = 'protobuf', schema.registry = 'http://wrong-host:8081')\n// after\nWITH (... format = 'plain', encode = 'protobuf', schema.registry = 'http://schema-registry:8081')","handlingStrategy":"try-catch","validationCode":"// Pre-check: schema subject exists in Confluent Schema Registry\ncurl -f ${SR_URL}/subjects/${topic}-value/latest || echo 'subject missing or registry unreachable';\n// Pre-check: inline proto compiles\nprotoc --descriptor_set_out=/dev/null your_schema.proto","typeGuard":null,"tryCatchPattern":"match sink_builder_result {\n    Err(e) if e.to_string().contains(\"fetch_descriptor\") || /* registry/proto errors */ true => {\n        // inspect inner message, fix proto syntax or registry config, then retry\n    }\n    r => r?,\n}","preventionTips":["Compile inline proto definitions with protoc before using them","Register the `<topic>-value` subject in Schema Registry ahead of sink creation","Verify Schema Registry URL and auth from the RisingWave nodes","Keep proto files version-controlled and validated in CI"],"tags":["protobuf","schema-registry","config","sink"],"backgroundTag":"schema-validation-failed","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"}