risingwavelabs/risingwave · error · SinkError::Config

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

Error message

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

What it means

During Kafka sink validation, any non-AppendOnly format (e.g. UPSERT/DEBEZIUM) requires a primary key so change records can be keyed correctly. If pk_indices is empty for such a sink, validate() returns this SinkError::Config naming the offending format.

Source

Thrown at src/connector/src/sink/kafka.rs:380

        .await?;
        let max_delivery_buffer_size = (self
            .config
            .rdkafka_properties_producer
            .queue_buffering_max_messages
            .as_ref()
            .cloned()
            .unwrap_or(KAFKA_WRITER_MAX_QUEUE_SIZE) as f32
            * KAFKA_WRITER_MAX_QUEUE_SIZE_RATIO) as usize;

        Ok(KafkaSinkWriter::new(self.config.clone(), formatter)
            .await?
            .into_log_sinker(max_delivery_buffer_size))
    }

    async fn validate(&self) -> Result<()> {
        // For non-append-only Kafka sink, the primary key must be defined.
        if self.format_desc.format != SinkFormat::AppendOnly && self.pk_indices.is_empty() {
            return Err(SinkError::Config(anyhow!(
                "primary key not defined for {:?} kafka 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.pk_indices.clone(),
            self.db_name.clone(),
            self.sink_from_name.clone(),
            &self.config.common.topic,
        )
        .await?;

        // Try Kafka connection.
        // There is no such interface for kafka producer to validate a connection
        // use enumerator to validate broker reachability and existence of topic

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add a `primary_key` definition to the CREATE SINK statement
2. Use FORMAT PLAIN (append-only) if no key is needed
3. Ensure the sinked relation actually has a primary key when relying on implicit propagation

Example fix

// before
CREATE SINK s FROM t WITH (connector='kafka', type='upsert', ...) FORMAT UPSERT ENCODE JSON;
// after
CREATE SINK s FROM t WITH (connector='kafka', type='upsert', primary_key='id', ...) FORMAT UPSERT ENCODE JSON;

fn kafka_pk_ok(format: SinkFormat, pk_indices: &[usize]) -> bool {
    format == SinkFormat::AppendOnly || !pk_indices.is_empty()
}

if let Err(e) = sink.validate().await {
    return Err(anyhow!("kafka sink validation failed: {e}"));
}

When it happens

Trigger: Calling Sink::validate() on a KafkaSink whose format is UPSERT or DEBEZIUM and whose CREATE SINK statement did not define a primary_key column list.

Common situations: Creating an upsert Kafka sink over a source/table without declaring `primary_key` in the WITH/options; sink over a table whose primary key was dropped; planner passing empty pk_indices.

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

• schemas.enable is expected to be `true` or `false`, got {s}
• serde (de)serialization error for KafkaConfig: {e}
• properties `scan_startup_mode` only supports earliest and la
• expected JSON in the form {{"host": "endpoint url"}}, but go
• unrecognized configs: {:?}

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