risingwavelabs/risingwave · error · SinkError::Kinesis

no key provided

Error message

no key provided

What it means

FormattedSink::write_one for Kinesis requires a key because Kinesis records must have a partition key. If the formatter upstream produces no key (Option<String> is None), the sink errors out instead of writing a record without a key.

Source

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

        let size = key.len() + payload.len();
        self.entries.push((
            PutRecordsRequestEntry::builder()
                .partition_key(key)
                .data(Blob::new(payload))
                .build()
                .expect("should not fail because we have set `data` and `partition_key`"),
            size,
        ))
    }
}

impl FormattedSink for KinesisSinkPayloadWriter {
    type K = String;
    type V = Vec<u8>;

    async fn write_one(&mut self, k: Option<Self::K>, v: Option<Self::V>) -> Result<()> {
        self.put_record(
            k.ok_or_else(|| SinkError::Kinesis(anyhow!("no key provided")))?,
            v.unwrap_or_default(),
        );
        Ok(())
    }
}

impl AsyncTruncateSinkWriter for KinesisSinkWriter {
    type DeliveryFuture = KinesisSinkPayloadWriterDeliveryFuture;

    async fn write_chunk<'a>(
        &'a mut self,
        chunk: StreamChunk,
        mut add_future: DeliveryFutureManagerAddFuture<'a, Self::DeliveryFuture>,
    ) -> Result<()> {
        let mut payload_writer = self.new_payload_writer();
        dispatch_sink_formatter_str_key_impl!(
            &self.formatter,
            formatter,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure the sink has a primary key so the formatter emits a key (see validate's partition-key requirement)
2. Use a format/encode combination that produces a key (e.g. `FORMAT UPSERT` or append-only with key columns)
3. If a record legitimately has no key, sink to a connector that does not require one

// ensure the sink formatter will emit a key
// upstream: MV/table must have primary key; sink must declare primary_key
CREATE SINK s FROM mv WITH (connector='kinesis', stream='s1',
  primary_key='id', format='upsert');

match sink.write_one(k, v).await {
  Err(e) if e.to_string().contains("no key provided") => {
    error!("formatter produced no key; check primary_key/format settings");
    Err(e)
  }
  r => r,
}

When it happens

Trigger: `write_one(k: None, v)` — a formatter/encoder path yields no key for the record, e.g. a format that does not extract a key column or a downstream formatter dropping the key.

Common situations: Using `FORMAT DEBEZIUM`/`UPSERT`-style encoders where the key is missing; sink formatter failing to map primary key columns; misconfigured format encode options.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• kinesis sink requires partition key (please define in `prima
• no value find in sink schema, index is 0
• no value find in sink schema, index is {:?}
• please set the separator in the with option, when there are
• Kinesis error: {0}

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