{"record":{"id":"e0ecf70ba6ced996","repo":"risingwavelabs/risingwave","slug":"missing-format-encode-e0ecf7","errorCode":null,"errorMessage":"missing FORMAT ... ENCODE ...","messagePattern":"missing FORMAT \\.\\.\\. ENCODE \\.\\.\\.","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/google_pubsub.rs","lineNumber":189,"sourceCode":"            &self.format_desc,\n            self.db_name.clone(),\n            self.sink_from_name.clone(),\n        )\n        .await?\n        .into_log_sinker(PUBSUB_SEND_FUTURE_BUFFER_MAX_SIZE))\n    }\n}\n\nimpl TryFrom<SinkParam> for GooglePubSubSink {\n    type Error = SinkError;\n\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        let schema = param.schema();\n        let pk_indices = param.downstream_pk_or_empty();\n        let config = GooglePubSubConfig::from_btreemap(param.properties)?;\n        let format_desc = param\n            .format_desc\n            .ok_or_else(|| SinkError::Config(anyhow!(\"missing FORMAT ... ENCODE ...\")))?;\n        Ok(Self {\n            config,\n            is_append_only: param.sink_type.is_append_only(),\n            schema,\n            pk_indices,\n            format_desc,\n            db_name: param.db_name,\n            sink_from_name: param.sink_from_name,\n        })\n    }\n}\n\nstruct GooglePubSubPayloadWriter<'w> {\n    publisher: &'w mut Publisher,\n    message_vec: Vec<PubsubMessage>,\n    add_future: DeliveryFutureManagerAddFuture<'w, GooglePubSubSinkDeliveryFuture>,\n}\n","sourceCodeStart":171,"sourceCodeEnd":207,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/google_pubsub.rs#L171-L207","documentation":"SinkFromParam::try_from expects the sink definition to carry a format descriptor produced by the `FORMAT ... ENCODE ...` clause. Google Pub/Sub sinks need it to know how to encode the payload. When param.format_desc is None (no format clause given), construction fails with this configuration error.","triggerScenarios":"`CREATE SINK ... WITH (connector='google_pubsub', ...)` omitted the `FORMAT APPEND ONLY ENCODE ...` clause, so SinkParam.format_desc is None when the sink is built.","commonSituations":"Users coming from older RisingWave syntax that only specified `connector` and `type` without FORMAT/ENCODE; copy-pasted sink DDL missing the format clause.","solutions":["Add the format clause: `FORMAT APPEND ONLY ENCODE JSON` (Pub/Sub only supports append-only)","Ensure the sink DDL includes both FORMAT and ENCODE keywords","If constructing SinkParam programmatically, populate format_desc before calling try_from"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector='google_pubsub');\n// after\nCREATE SINK s FROM mv WITH (connector='google_pubsub', FORMAT APPEND ONLY ENCODE JSON);","handlingStrategy":"validation","validationCode":"fn ensure_format_clause(ddl: &str) -> Result<(), String> {\n    if !ddl.to_uppercase().contains(\"FORMAT APPEND ONLY ENCODE\") {\n        return Err(\"google_pubsub sinks require FORMAT APPEND ONLY ENCODE ...\".into());\n    }\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never omit the FORMAT...ENCODE clause for connector sinks","Use sink DDL templates that include the format clause","When building SinkParam in code, always populate format_desc"],"tags":["pubsub","sink","config"],"backgroundTag":"missing-required-config-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"}