risingwavelabs/risingwave · error · SinkError

{e}

Error message

{e}

What it means

When building a protobuf encoder, the connector fetches the `.proto` file descriptor (from provided options or the Confluent Schema Registry, using the topic as subject). Any failure in `fetch_descriptor` — unparseable proto content, unreachable registry, missing subject — is wrapped into this Config error with the underlying message.

Source

Thrown at src/connector/src/sink/formatter/mod.rs:178

                    )));
                }
            }
        } else {
            encoder
        };
        Ok(encoder)
    }
}

impl EncoderBuild for ProtoEncoder {
    async fn build(b: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {
        // TODO: better to be a compile-time assert
        assert!(pk_indices.is_none());
        // By passing `None` as `aws_auth_props`, reading from `s3://` not supported yet.
        let (descriptor, sid) =
            crate::schema::protobuf::fetch_descriptor(&b.format_desc.options, b.topic, None)
                .await
                .map_err(|e| SinkError::Config(anyhow!(e)))?;
        let header = match sid {
            None => ProtoHeader::None,
            Some(sid) => ProtoHeader::ConfluentSchemaRegistry(sid),
        };
        ProtoEncoder::new(b.schema, None, descriptor, header)
    }
}

fn ensure_only_one_pk<'a>(
    data_type_name: &'a str,
    params: &'a EncoderParams<'_>,
    pk_indices: &'a Option<Vec<usize>>,
) -> Result<(usize, &'a Field)> {
    let Some(pk_indices) = pk_indices else {
        return Err(SinkError::Config(anyhow!(
            "{}Encoder requires primary key columns to be specified",
            data_type_name
        )));

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the underlying `{e}` message: fix proto syntax if inline, or registry URL/auth if fetched remotely
2. Register or correct the schema subject for the topic in Confluent Schema Registry
3. Validate the inline proto compiles (e.g. with protoc) before using it in the sink
4. Verify network reachability to the Schema Registry from RisingWave nodes

Example fix

// before
WITH (... format = 'plain', encode = 'protobuf', schema.registry = 'http://wrong-host:8081')
// after
WITH (... format = 'plain', encode = 'protobuf', schema.registry = 'http://schema-registry:8081')

// Pre-check: schema subject exists in Confluent Schema Registry
curl -f ${SR_URL}/subjects/${topic}-value/latest || echo 'subject missing or registry unreachable';
// Pre-check: inline proto compiles
protoc --descriptor_set_out=/dev/null your_schema.proto

match sink_builder_result {
    Err(e) if e.to_string().contains("fetch_descriptor") || /* registry/proto errors */ true => {
        // inspect inner message, fix proto syntax or registry config, then retry
    }
    r => r?,
}

When it happens

Trigger: Building a protobuf-encoded sink where the descriptor cannot be fetched: inline proto option has syntax errors, Schema Registry URL/credentials are wrong, or no schema exists for the topic.

Common situations: Typo'd proto definitions, Schema Registry down or auth-protected, subject naming mismatch (`<topic>-value` not registered), and lack of registry connectivity from compute nodes.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• unused FORMAT ENCODE option: key.message='{name}'
• protobuf file location not provided
• Must specify '{}' or '{}'
• Can't find message proto {}
• sink type unsupported: {}

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