risingwavelabs/risingwave · error · SinkError::Config

missing FORMAT ... ENCODE ...

Error message

missing FORMAT ... ENCODE ...

What it means

The Pulsar sink constructor requires a FORMAT ... ENCODE declaration (parsed into param.format_desc) to know how to serialize rows and attach keys/schemas. When the SinkParam has no format description, the constructor returns SinkError::Config with this message. The library throws it because a Pulsar sink cannot determine its data encoding without FORMAT/ENCODE.

Source

Thrown at src/connector/src/sink/pulsar.rs:242

        }
        Ok(())
    }
}

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

    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
        let schema = param.schema();
        let downstream_pk = param.downstream_pk_or_empty();
        let config = PulsarConfig::from_btreemap(param.properties)?;
        Ok(Self {
            config,
            schema,
            downstream_pk,
            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 PulsarSink {
    type LogSinker = AsyncTruncateLogSinkerOf<PulsarSinkWriter>;

    const SINK_NAME: &'static str = PULSAR_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        // Reduce async state machine size (see `clippy::large_futures`).
        let writer = Box::pin(PulsarSinkWriter::new(
            self.config.clone(),
            self.schema.clone(),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add a FORMAT ... ENCODE ... clause to the CREATE SINK statement, e.g. FORMAT DEBEZIUM ENCODE JSON or FORMAT APPEND ONLY ENCODE JSON.
2. Ensure the sink is created through the SQL path that populates SinkParam::format_desc.
3. If creating sinks programmatically, construct SinkFormatDesc and set it on SinkParam before building the Pulsar sink.
4. Check RisingWave docs for supported FORMAT/ENCODE combinations for the Pulsar connector.

Example fix

// before
CREATE SINK s FROM mv WITH (connector = 'pulsar', service.url = '...', topic = 't');
// after
CREATE SINK s FROM mv WITH (connector = 'pulsar', service.url = '...', topic = 't')
FORMAT DEBEZIUM ENCODE JSON;

// Check the DDL carries FORMAT ... ENCODE before creating a pulsar sink
let ok = ddl.to_lowercase().contains("format") && ddl.to_lowercase().contains("encode");
if !ok { return Err("pulsar sink requires FORMAT ... ENCODE ... clause".into()); }

fn has_format_desc(param: &SinkParam) -> bool { param.format_desc.is_some() }

match PulsarSink::new(param, ...) {
    Err(SinkError::Config(e)) if e.to_string().contains("missing FORMAT") => {
        eprintln!("add FORMAT ... ENCODE ... to the CREATE SINK statement");
    }
    other => other?,
}

When it happens

Trigger: Creating a Pulsar sink via SinkParam::new without a FORMAT..ENCODE clause in the DDL, e.g. CREATE SINK ... WITH (connector='pulsar', ...) lacking 'format appendonly encode json' or 'format debezium encode avro'.

Common situations: Users forgetting the FORMAT/ENCODE clause in the CREATE SINK statement; programmatic sink creation that omits format_desc; copying an example DDL that dropped the format line.

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

• {serde_json deserialization error for PulsarConfig from prop
• missing FORMAT ... ENCODE ...
• `{}` must be {}, or {}
• Primary key not defined for upsert doris sink (please define
• Can't get fe host from url

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