risingwavelabs/risingwave · error · SinkError::Config

missing FORMAT ... ENCODE ...

Error message

missing FORMAT ... ENCODE ...

What it means

KinesisSink construction requires a parsed format descriptor (FORMAT ... ENCODE ...) from the sink DDL; if param.format_desc is None it fails with this SinkError::Config because the sink cannot determine the record serialization format.

Source

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

        }
        Ok(())
    }
}

impl TryFrom<SinkParam> for KinesisSink {
    type Error = SinkError;

    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
        let schema = param.schema();
        let pk_indices = param.downstream_pk_or_empty();
        let config = KinesisSinkConfig::from_btreemap(param.properties)?;
        Ok(Self {
            config,
            schema,
            pk_indices,
            format_desc: param
                .format_desc
                .ok_or_else(|| SinkError::Config(anyhow!("missing FORMAT ... ENCODE ...")))?,
            db_name: param.db_name,
            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

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Add FORMAT ... ENCODE ... to the CREATE SINK statement (e.g. FORMAT PLAIN ENCODE JSON)
  2. Confirm the exact statement syntax parses the format before sink construction
  3. When building SinkParam programmatically, set format_desc explicitly

Example fix

// before
CREATE SINK s FROM t WITH (connector='kinesis', stream='my-stream');
// after
CREATE SINK s FROM t WITH (connector='kinesis', stream='my-stream') FORMAT PLAIN ENCODE JSON;
Defensive patterns

Strategy: validation

Validate before calling

fn ensure_kinesis_format(param: &SinkParam) -> Result<()> {
    if param.format_desc.is_none() { bail!("kinesis sink requires FORMAT ... ENCODE ..."); }
    Ok(())
}

Try / catch

let format_desc = param.format_desc.as_ref()
    .ok_or_else(|| anyhow!("kinesis sink created without FORMAT ... ENCODE ..."))?;

Prevention

When it happens

Trigger: Creating a Kinesis sink without FORMAT ... ENCODE ... clauses in CREATE SINK, or constructing a KinesisSink from a SinkParam whose format_desc was never populated.

Common situations: Omitting FORMAT/ENCODE in the CREATE SINK statement for Kinesis; hand-built SinkParam in code or tests missing format_desc; syntax mistakes causing the parser to skip format parsing.

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

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