{"record":{"id":"27ff8a1117bc98fd","repo":"risingwavelabs/risingwave","slug":"kinesis-sink-requires-partition-key-please-define","errorCode":null,"errorMessage":"kinesis sink requires partition key (please define in `primary_key` field)","messagePattern":"kinesis sink requires partition key \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":97,"sourceCode":"            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\n        if self.pk_indices.is_empty() {\n            return Err(SinkError::Config(anyhow!(\n                \"kinesis sink requires partition key (please define in `primary_key` field)\",\n            )));\n        }\n        // Check for formatter constructor error, before it is too late for error reporting.\n        SinkFormatterImpl::new(\n            &self.format_desc,\n            self.schema.clone(),\n            self.pk_indices.clone(),\n            self.db_name.clone(),\n            self.sink_from_name.clone(),\n            &self.config.common.stream_name,\n        )\n        .await?;\n\n        // check reachability\n        let client = self.config.common.build_client().await?;\n        client\n            .list_shards()","sourceCodeStart":79,"sourceCodeEnd":115,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L79-L115","documentation":"Kinesis requires every record to carry a partition key, which RisingWave derives from the sink's primary key columns. The sink validator rejects a Kinesis sink whose pk_indices is empty because there is no builtin round-robin partitioning fallback as there is for kafka/pulsar sinks.","triggerScenarios":"Creating a Kinesis sink whose defining query or MV has no primary key, so `pk_indices.is_empty()` during `validate`.","commonSituations":"Sinking a source without a PRIMARY KEY clause; forgetting `primary_key` in the WITH options; assuming Kinesis supports round-robin like the Kafka sink.","solutions":["Define a `primary_key` for the sink (e.g. `CREATE SINK ... AS SELECT ... WITH ( primary_key = 'col1,col2' )` or ensure the upstream MV/table has a primary key)","Add a row identifier (e.g. row_id column) to the query so a key exists","If keyless shipping is truly desired, use a sink type that supports round-robin partitioning (kafka/pulsar) instead of Kinesis"],"exampleFix":"// before\nCREATE SINK k_sink FROM mv WITH (\n  connector = 'kinesis', stream = 's1'\n);\n// after\nCREATE SINK k_sink FROM mv WITH (\n  connector = 'kinesis', stream = 's1', primary_key = 'user_id'\n);","handlingStrategy":"validation","validationCode":"// SQL-side check before CREATE SINK\n-- ensure the upstream relation has a primary key:\nSHOW COLUMNS FROM mv; -- or query rw_catalog for primary key\n-- must define WITH ( primary_key = '...' ) for keyless sources","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always specify primary_key in WITH options for Kinesis sinks","Model upstream data as a table/MV with a PRIMARY KEY","Prefer kafka/pulsar sinks when keyless round-robin delivery is intended"],"tags":["kinesis","sink","partition-key","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"}