{"record":{"id":"2e82db68e216bf0a","repo":"risingwavelabs/risingwave","slug":"missing-format-encode-2e82db","errorCode":null,"errorMessage":"missing FORMAT ... ENCODE ...","messagePattern":"missing FORMAT \\.\\.\\. ENCODE \\.\\.\\.","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/pulsar.rs","lineNumber":242,"sourceCode":"        }\n        Ok(())\n    }\n}\n\nimpl TryFrom<SinkParam> for PulsarSink {\n    type Error = SinkError;\n\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        let schema = param.schema();\n        let downstream_pk = param.downstream_pk_or_empty();\n        let config = PulsarConfig::from_btreemap(param.properties)?;\n        Ok(Self {\n            config,\n            schema,\n            downstream_pk,\n            format_desc: param\n                .format_desc\n                .ok_or_else(|| SinkError::Config(anyhow!(\"missing FORMAT ... ENCODE ...\")))?,\n            db_name: param.db_name,\n            sink_from_name: param.sink_from_name,\n        })\n    }\n}\n\nimpl Sink for PulsarSink {\n    type LogSinker = AsyncTruncateLogSinkerOf<PulsarSinkWriter>;\n\n    const SINK_NAME: &'static str = PULSAR_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {\n        // Reduce async state machine size (see `clippy::large_futures`).\n        let writer = Box::pin(PulsarSinkWriter::new(\n            self.config.clone(),\n            self.schema.clone(),","sourceCodeStart":224,"sourceCodeEnd":260,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/pulsar.rs#L224-L260","documentation":"The Pulsar sink constructor requires a FORMAT ... ENCODE declaration (parsed into param.format_desc) to know how to serialize rows and attach keys/schemas. When the SinkParam has no format description, the constructor returns SinkError::Config with this message. The library throws it because a Pulsar sink cannot determine its data encoding without FORMAT/ENCODE.","triggerScenarios":"Creating a Pulsar sink via SinkParam::new without a FORMAT..ENCODE clause in the DDL, e.g. CREATE SINK ... WITH (connector='pulsar', ...) lacking 'format appendonly encode json' or 'format debezium encode avro'.","commonSituations":"Users forgetting the FORMAT/ENCODE clause in the CREATE SINK statement; programmatic sink creation that omits format_desc; copying an example DDL that dropped the format line.","solutions":["Add a FORMAT ... ENCODE ... clause to the CREATE SINK statement, e.g. FORMAT DEBEZIUM ENCODE JSON or FORMAT APPEND ONLY ENCODE JSON.","Ensure the sink is created through the SQL path that populates SinkParam::format_desc.","If creating sinks programmatically, construct SinkFormatDesc and set it on SinkParam before building the Pulsar sink.","Check RisingWave docs for supported FORMAT/ENCODE combinations for the Pulsar connector."],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector = 'pulsar', service.url = '...', topic = 't');\n// after\nCREATE SINK s FROM mv WITH (connector = 'pulsar', service.url = '...', topic = 't')\nFORMAT DEBEZIUM ENCODE JSON;","handlingStrategy":"validation","validationCode":"// Check the DDL carries FORMAT ... ENCODE before creating a pulsar sink\nlet ok = ddl.to_lowercase().contains(\"format\") && ddl.to_lowercase().contains(\"encode\");\nif !ok { return Err(\"pulsar sink requires FORMAT ... ENCODE ... clause\".into()); }","typeGuard":"fn has_format_desc(param: &SinkParam) -> bool { param.format_desc.is_some() }","tryCatchPattern":"match PulsarSink::new(param, ...) {\n    Err(SinkError::Config(e)) if e.to_string().contains(\"missing FORMAT\") => {\n        eprintln!(\"add FORMAT ... ENCODE ... to the CREATE SINK statement\");\n    }\n    other => other?,\n}","preventionTips":["Always end CREATE SINK for pulsar with FORMAT .. ENCODE ..","Use DEBEZIUM ENCODE JSON or APPEND ONLY ENCODE JSON","Review generated DDL templates for the format line","Programmatic sink creation: populate SinkParam::format_desc"],"tags":["rust","pulsar","sink","config","sql"],"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"}