{"record":{"id":"4e2798fc0ece8d56","repo":"risingwavelabs/risingwave","slug":"missing-format-encode-4e2798","errorCode":null,"errorMessage":"missing FORMAT ... ENCODE ...","messagePattern":"missing FORMAT \\.\\.\\. ENCODE \\.\\.\\.","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":77,"sourceCode":"        }\n        Ok(())\n    }\n}\n\nimpl TryFrom<SinkParam> for KinesisSink {\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 = KinesisSinkConfig::from_btreemap(param.properties)?;\n        Ok(Self {\n            config,\n            schema,\n            pk_indices,\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\nconst KINESIS_SINK_MAX_PENDING_CHUNK_NUM: usize = 64;\n\nimpl Sink for KinesisSink {\n    type LogSinker = AsyncTruncateLogSinkerOf<KinesisSinkWriter>;\n\n    const SINK_NAME: &'static str = KINESIS_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        // Kinesis requires partition key. There is no builtin support for round-robin as in kafka/pulsar.\n        // https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#Streams-PutRecord-request-PartitionKey","sourceCodeStart":59,"sourceCodeEnd":95,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L59-L95","documentation":"KinesisSink construction requires a parsed format descriptor (FORMAT ... ENCODE ...) from the sink DDL; if param.format_desc is None it fails with this SinkError::Config because the sink cannot determine the record serialization format.","triggerScenarios":"Creating a Kinesis sink without FORMAT ... ENCODE ... clauses in CREATE SINK, or constructing a KinesisSink from a SinkParam whose format_desc was never populated.","commonSituations":"Omitting FORMAT/ENCODE in the CREATE SINK statement for Kinesis; hand-built SinkParam in code or tests missing format_desc; syntax mistakes causing the parser to skip format parsing.","solutions":["Add FORMAT ... ENCODE ... to the CREATE SINK statement (e.g. FORMAT PLAIN ENCODE JSON)","Confirm the exact statement syntax parses the format before sink construction","When building SinkParam programmatically, set format_desc explicitly"],"exampleFix":"// before\nCREATE SINK s FROM t WITH (connector='kinesis', stream='my-stream');\n// after\nCREATE SINK s FROM t WITH (connector='kinesis', stream='my-stream') FORMAT PLAIN ENCODE JSON;","handlingStrategy":"validation","validationCode":"fn ensure_kinesis_format(param: &SinkParam) -> Result<()> {\n    if param.format_desc.is_none() { bail!(\"kinesis sink requires FORMAT ... ENCODE ...\"); }\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":"let format_desc = param.format_desc.as_ref()\n    .ok_or_else(|| anyhow!(\"kinesis sink created without FORMAT ... ENCODE ...\"))?;","preventionTips":["Always append FORMAT ... ENCODE ... to Kinesis CREATE SINK statements","Keep a validated DDL template for kinesis sinks","Parse/test the statement before applying it in production"],"tags":["kinesis","config","format","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"}