risingwavelabs/risingwave · error · SinkError::Config

missing FORMAT ... ENCODE ...

Error message

missing FORMAT ... ENCODE ...

What it means

KafkaSink is constructed from SinkParam and requires a parsed format descriptor (FORMAT ... ENCODE ... from the DDL). If param.format_desc is None, construction fails with this SinkError::Config, because the sink cannot know how to serialize messages without a declared format/encode.

Source

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

        }
        Ok(())
    }
}

impl TryFrom<SinkParam> for KafkaSink {
    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 = KafkaConfig::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,
        })
    }
}

impl Sink for KafkaSink {
    type LogSinker = AsyncTruncateLogSinkerOf<KafkaSinkWriter>;

    const SINK_NAME: &'static str = KAFKA_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        let formatter = SinkFormatterImpl::new(
            &self.format_desc,
            self.schema.clone(),
            self.pk_indices.clone(),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add FORMAT ... ENCODE ... to the CREATE SINK statement (e.g. FORMAT PLAIN ENCODE JSON)
2. Verify the statement parses the format by checking desc/param before sink creation
3. When building SinkParam in code, populate format_desc from parsed options

Example fix

// before
CREATE SINK s FROM t WITH (connector='kafka', properties.bootstrap.server='b:9092');
// after
CREATE SINK s FROM t WITH (connector='kafka', properties.bootstrap.server='b:9092') FORMAT PLAIN ENCODE JSON;

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

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

When it happens

Trigger: Creating a Kafka sink whose DDL lacks FORMAT ... ENCODE ... clauses, or an internal path building KafkaSink from a SinkParam where format_desc was never parsed (e.g. malformed or missing FORMAT/ENCODE in CREATE SINK).

Common situations: Writing CREATE SINK statements without `FORMAT PLAIN/DEBEZIUM/UPSERT ENCODE JSON/AVRO/...`; constructing sinks programmatically with a hand-built SinkParam; upgrading from older syntax without format declarations.

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

• missing FORMAT ... ENCODE ...
• Must specify '{}' or '{}'
• sink format unsupported: {}
• missing FORMAT ... ENCODE ...
• schemas.enable is expected to be `true` or `false`, got {s}

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