{"record":{"id":"f4f2c5165c0131cb","repo":"risingwavelabs/risingwave","slug":"no-key-provided","errorCode":null,"errorMessage":"no key provided","messagePattern":"no key provided","errorType":"exception","errorClass":"SinkError::Kinesis","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":422,"sourceCode":"        let size = key.len() + payload.len();\n        self.entries.push((\n            PutRecordsRequestEntry::builder()\n                .partition_key(key)\n                .data(Blob::new(payload))\n                .build()\n                .expect(\"should not fail because we have set `data` and `partition_key`\"),\n            size,\n        ))\n    }\n}\n\nimpl FormattedSink for KinesisSinkPayloadWriter {\n    type K = String;\n    type V = Vec<u8>;\n\n    async fn write_one(&mut self, k: Option<Self::K>, v: Option<Self::V>) -> Result<()> {\n        self.put_record(\n            k.ok_or_else(|| SinkError::Kinesis(anyhow!(\"no key provided\")))?,\n            v.unwrap_or_default(),\n        );\n        Ok(())\n    }\n}\n\nimpl AsyncTruncateSinkWriter for KinesisSinkWriter {\n    type DeliveryFuture = KinesisSinkPayloadWriterDeliveryFuture;\n\n    async fn write_chunk<'a>(\n        &'a mut self,\n        chunk: StreamChunk,\n        mut add_future: DeliveryFutureManagerAddFuture<'a, Self::DeliveryFuture>,\n    ) -> Result<()> {\n        let mut payload_writer = self.new_payload_writer();\n        dispatch_sink_formatter_str_key_impl!(\n            &self.formatter,\n            formatter,","sourceCodeStart":404,"sourceCodeEnd":440,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L404-L440","documentation":"FormattedSink::write_one for Kinesis requires a key because Kinesis records must have a partition key. If the formatter upstream produces no key (Option<String> is None), the sink errors out instead of writing a record without a key.","triggerScenarios":"`write_one(k: None, v)` — a formatter/encoder path yields no key for the record, e.g. a format that does not extract a key column or a downstream formatter dropping the key.","commonSituations":"Using `FORMAT DEBEZIUM`/`UPSERT`-style encoders where the key is missing; sink formatter failing to map primary key columns; misconfigured format encode options.","solutions":["Ensure the sink has a primary key so the formatter emits a key (see validate's partition-key requirement)","Use a format/encode combination that produces a key (e.g. `FORMAT UPSERT` or append-only with key columns)","If a record legitimately has no key, sink to a connector that does not require one"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// ensure the sink formatter will emit a key\n// upstream: MV/table must have primary key; sink must declare primary_key\nCREATE SINK s FROM mv WITH (connector='kinesis', stream='s1',\n  primary_key='id', format='upsert');","typeGuard":null,"tryCatchPattern":"match sink.write_one(k, v).await {\n  Err(e) if e.to_string().contains(\"no key provided\") => {\n    error!(\"formatter produced no key; check primary_key/format settings\");\n    Err(e)\n  }\n  r => r,\n}","preventionTips":["Always configure primary_key for Kinesis sinks","Use FORMAT UPSERT or key-capable encoders","Verify the chosen formatter emits keys for append-only streams too"],"tags":["kinesis","sink","partition-key","formatter"],"backgroundTag":"missing-required-argument","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"}