risingwavelabs/risingwave · error · SinkError::Config

kinesis sink requires partition key (please define in `prima

Error message

kinesis sink requires partition key (please define in `primary_key` field)

What it means

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.

Source

Thrown at src/connector/src/sink/kinesis.rs:97

            sink_from_name: param.sink_from_name,
        })
    }
}

const KINESIS_SINK_MAX_PENDING_CHUNK_NUM: usize = 64;

impl Sink for KinesisSink {
    type LogSinker = AsyncTruncateLogSinkerOf<KinesisSinkWriter>;

    const SINK_NAME: &'static str = KINESIS_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn validate(&self) -> Result<()> {
        // Kinesis requires partition key. There is no builtin support for round-robin as in kafka/pulsar.
        // https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#Streams-PutRecord-request-PartitionKey
        if self.pk_indices.is_empty() {
            return Err(SinkError::Config(anyhow!(
                "kinesis sink requires partition key (please define in `primary_key` field)",
            )));
        }
        // 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.stream_name,
        )
        .await?;

        // check reachability
        let client = self.config.common.build_client().await?;
        client
            .list_shards()

View on GitHub (pinned to 6469eb736d)

Solutions

1. 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)
2. Add a row identifier (e.g. row_id column) to the query so a key exists
3. If keyless shipping is truly desired, use a sink type that supports round-robin partitioning (kafka/pulsar) instead of Kinesis

Example fix

// before
CREATE SINK k_sink FROM mv WITH (
  connector = 'kinesis', stream = 's1'
);
// after
CREATE SINK k_sink FROM mv WITH (
  connector = 'kinesis', stream = 's1', primary_key = 'user_id'
);

// SQL-side check before CREATE SINK
-- ensure the upstream relation has a primary key:
SHOW COLUMNS FROM mv; -- or query rw_catalog for primary key
-- must define WITH ( primary_key = '...' ) for keyless sources

When it happens

Trigger: Creating a Kinesis sink whose defining query or MV has no primary key, so `pk_indices.is_empty()` during `validate`.

Common situations: Sinking a source without a PRIMARY KEY clause; forgetting `primary_key` in the WITH options; assuming Kinesis supports round-robin like the Kafka sink.

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

• no key provided
• sink type unsupported: {}
• sink format unsupported: {}
• sink encode unsupported: {}
• unsupported {} as sink key encode

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