{"record":{"id":"9f0fe968f4bfa362","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-pulsar-sink-plea","errorCode":null,"errorMessage":"primary key not defined for {:?} pulsar sink (please define in `primary_key` field)","messagePattern":"primary key not defined for (.+?) pulsar sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/pulsar.rs","lineNumber":273,"sourceCode":"\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(),\n            self.downstream_pk.clone(),\n            &self.format_desc,\n            self.db_name.clone(),\n            self.sink_from_name.clone(),\n        ))\n        .await?;\n        Ok(writer.into_log_sinker(PULSAR_SEND_FUTURE_BUFFER_MAX_SIZE))\n    }\n\n    async fn validate(&self) -> Result<()> {\n        // For non-append-only Pulsar sink, the primary key must be defined.\n        if self.format_desc.format != SinkFormat::AppendOnly && self.downstream_pk.is_empty() {\n            return Err(SinkError::Config(anyhow!(\n                \"primary key not defined for {:?} pulsar sink (please define in `primary_key` field)\",\n                self.format_desc.format\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.downstream_pk.clone(),\n            self.db_name.clone(),\n            self.sink_from_name.clone(),\n            &self.config.common.topic,\n        )\n        .await?;\n\n        // Validate pulsar connection.\n        let pulsar = self\n            .config","sourceCodeStart":255,"sourceCodeEnd":291,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/pulsar.rs#L255-L291","documentation":"Pulsar is a message bus without enforced keys, so for non-append-only sinks RisingWave requires an explicit primary key to compute the Pulsar message key and route/upsert records correctly. validate() returns SinkError::Config when the sink format is not AppendOnly and downstream_pk is empty. The library throws it to fail fast at sink creation instead of producing un-routable messages at runtime.","triggerScenarios":"Running validate() on a Pulsar sink whose format is Debezium (or any non-append-only format) while the DDL omitted a `primary_key` field in WITH options, leaving downstream_pk empty.","commonSituations":"Creating an upsert/Debezium Pulsar sink without `primary_key='...'` in WITH options; using a materialized view with a PK but not declaring the sink primary key; migrating sinks from append-only to upsert mode without adding the key.","solutions":["Add `primary_key` to the sink WITH options, naming the column(s) to use as the Pulsar message key.","Alternatively, if the sink truly only needs append semantics, change FORMAT to APPEND ONLY so the PK requirement is lifted.","Re-create the sink after adding the primary key field (sink options are fixed at creation).","Ensure the primary key columns exist in the sink's output schema and match the downstream consumer's keying expectations."],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector='pulsar', service.url='...', topic='t', type='upsert') FORMAT DEBEZIUM ENCODE JSON;\n// after\nCREATE SINK s FROM mv WITH (connector='pulsar', service.url='...', topic='t', type='upsert', primary_key='id') FORMAT DEBEZIUM ENCODE JSON;","handlingStrategy":"validation","validationCode":"// Guard before creating a non-append-only pulsar sink\nif format != SinkFormat::AppendOnly && (param.downstream_pk.as_ref().map_or(true, |pk| pk.is_empty())) {\n    return Err(\"non-append-only pulsar sink requires primary_key in WITH options\".into());\n}","typeGuard":"fn pk_defined(param: &SinkParam) -> bool {\n    param.downstream_pk.as_ref().map_or(false, |pk| !pk.is_empty())\n}","tryCatchPattern":"if let Err(SinkError::Config(e)) = sink.validate().await {\n    if e.to_string().contains(\"primary key not defined\") {\n        eprintln!(\"re-create the sink with primary_key in WITH options\");\n    }\n}","preventionTips":["Add primary_key whenever the pulsar sink is upsert/debezium","Match primary_key to the MV's primary key columns","Only omit primary_key for FORMAT APPEND ONLY","Validate sinks with `validate` before activation"],"tags":["rust","pulsar","sink","primary-key","validation"],"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"}