risingwavelabs/risingwave · error · SinkError::Config

primary key not defined for {:?} pulsar sink (please define

Error message

primary key not defined for {:?} pulsar sink (please define in `primary_key` field)

What it means

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.

Source

Thrown at src/connector/src/sink/pulsar.rs:273

    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        // Reduce async state machine size (see `clippy::large_futures`).
        let writer = Box::pin(PulsarSinkWriter::new(
            self.config.clone(),
            self.schema.clone(),
            self.downstream_pk.clone(),
            &self.format_desc,
            self.db_name.clone(),
            self.sink_from_name.clone(),
        ))
        .await?;
        Ok(writer.into_log_sinker(PULSAR_SEND_FUTURE_BUFFER_MAX_SIZE))
    }

    async fn validate(&self) -> Result<()> {
        // For non-append-only Pulsar sink, the primary key must be defined.
        if self.format_desc.format != SinkFormat::AppendOnly && self.downstream_pk.is_empty() {
            return Err(SinkError::Config(anyhow!(
                "primary key not defined for {:?} pulsar sink (please define in `primary_key` field)",
                self.format_desc.format
            )));
        }
        // Check for formatter constructor error, before it is too late for error reporting.
        SinkFormatterImpl::new(
            &self.format_desc,
            self.schema.clone(),
            self.downstream_pk.clone(),
            self.db_name.clone(),
            self.sink_from_name.clone(),
            &self.config.common.topic,
        )
        .await?;

        // Validate pulsar connection.
        let pulsar = self
            .config

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add `primary_key` to the sink WITH options, naming the column(s) to use as the Pulsar message key.
2. Alternatively, if the sink truly only needs append semantics, change FORMAT to APPEND ONLY so the PK requirement is lifted.
3. Re-create the sink after adding the primary key field (sink options are fixed at creation).
4. Ensure the primary key columns exist in the sink's output schema and match the downstream consumer's keying expectations.

Example fix

// before
CREATE SINK s FROM mv WITH (connector='pulsar', service.url='...', topic='t', type='upsert') FORMAT DEBEZIUM ENCODE JSON;
// after
CREATE SINK s FROM mv WITH (connector='pulsar', service.url='...', topic='t', type='upsert', primary_key='id') FORMAT DEBEZIUM ENCODE JSON;

// Guard before creating a non-append-only pulsar sink
if format != SinkFormat::AppendOnly && (param.downstream_pk.as_ref().map_or(true, |pk| pk.is_empty())) {
    return Err("non-append-only pulsar sink requires primary_key in WITH options".into());
}

fn pk_defined(param: &SinkParam) -> bool {
    param.downstream_pk.as_ref().map_or(false, |pk| !pk.is_empty())
}

if let Err(SinkError::Config(e)) = sink.validate().await {
    if e.to_string().contains("primary key not defined") {
        eprintln!("re-create the sink with primary_key in WITH options");
    }
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• Redis Sink Primary Key must be specified.
• Primary key not defined for upsert doris sink (please define
• DynamoDB table {} primary key {:?} must match RisingWave pri
• no value find in sink schema, index is {:?}
• please set the separator in the with option, when there are

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/9f0fe968f4bfa362. Report an issue: GitHub.