{"record":{"id":"b35d08c7ae9da5e7","repo":"risingwavelabs/risingwave","slug":"should-not-fail-because-we-have-set-data-and-pa","errorCode":null,"errorMessage":"should not fail because we have set `data` and `partition_key`","messagePattern":"should not fail because we have set `data` and `partition_key`","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":410,"sourceCode":"        //\n        // An unsuccessfully processed record includes ErrorCode and ErrorMessage values. ErrorCode reflects\n        // the type of error and can be one of the following values: ProvisionedThroughputExceededException or\n        // InternalFailure. ErrorMessage provides more detailed information about the ProvisionedThroughputExceededException\n        // exception including the account ID, stream name, and shard ID of the record that was throttled.\n        output\n            .records\n            .into_iter()\n            .find_position(|entry| entry.shard_id.is_none())\n    }\n\n    fn put_record(&mut self, key: String, payload: Vec<u8>) {\n        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","sourceCodeStart":392,"sourceCodeEnd":428,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L392-L428","documentation":"Building a PutRecordsRequestEntry can only fail if the required `data` or `partition_key` fields are unset. The sink always sets both, so it unwraps with expect(). This is an internal invariant violation, not a user-triggerable error.","triggerScenarios":"`put_record` in KinesisSinkWriter when `PutRecordsRequestEntry::builder().partition_key(key).data(Blob::new(payload)).build()` unexpectedly returns None.","commonSituations":"Only possible due to an SDK builder regression or code change that stops setting data/partition_key; never hit by end users in normal operation.","solutions":["If it panics, it is a bug in the sink or aws-sdk-kinesis version — file an issue and check SDK changelog","Pin/upgrade aws-sdk-kinesis to a version where the builder still requires these fields","As a code fix, replace expect() with a proper SinkError mapping for robustness"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// treat a panic here as a bug; catch at the task boundary\nmatch tokio::spawn(sink_task).await {\n  Ok(Err(e)) => error!(\"sink failed: {e:#}\"),\n  Err(join_err) => error!(\"sink task panicked: {join_err}\"), // covers expect()\n  Ok(Ok(_)) => {}\n}","preventionTips":["Keep aws-sdk-kinesis versions pinned and tested in CI","Never bypass put_record's builder (always set both data and partition_key)"],"tags":["kinesis","invariant","aws-sdk","assert"],"backgroundTag":"internal-invariant-violation","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"}