risingwavelabs/risingwave · error · SinkError::Config

Invalid metadata format

Error message

Invalid metadata format

What it means

Each sink metadata entry for the S3-manifest path must be the `Serialized` variant containing the written file path. This error is thrown when the metadata enum is any other variant (or None), so no file path can be extracted for the manifest.

Solutions

1. Ensure the source of the metadata populates `Metadata::Serialized` with the S3 path
2. Check RisingWave version consistency across frontend/compute/meta nodes
3. Replay the epoch with correctly formed metadata

// reject non-Serialized metadata upstream before commit
matches!(meta.metadata, Some(Metadata::Serialized(_)));

fn is_serialized(m: &Option<SinkMetadataEnum>) -> bool {
    matches!(m, Some(Metadata::Serialized(_)))
}

When it happens

Trigger: `commit_data` receives `SinkMetadata` whose `metadata` field is not `sink_metadata::Metadata::Serialized(...)` — e.g., a different metadata type or missing metadata.

Common situations: Downstream/upstream version mismatch in the sink metadata protobuf; users writing custom metadata into the stream; internal bug populating the wrong metadata variant.

Related errors

• (dynamic: UTF-8 conversion error from metadata bytes)
• Can't create iceberg sink write result from empty data!
• (dynamic: channel send error on shutdown signal)
• {e}
• iceberg sink metadata should be an object

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

Appendix: source

Thrown at src/connector/src/sink/snowflake_redshift/redshift.rs:676

        let paths = metadata
            .into_iter()
            .filter(|m| {
                if let Some(sink_metadata::Metadata::Serialized(SerializedMetadata { metadata })) =
                    &m.metadata
                {
                    !metadata.is_empty()
                } else {
                    false
                }
            })
            .map(|metadata| {
                let path = if let Some(sink_metadata::Metadata::Serialized(SerializedMetadata {
                    metadata,
                })) = metadata.metadata
                {
                    String::from_utf8(metadata).map_err(|e| SinkError::Config(anyhow!(e)))
                } else {
                    Err(SinkError::Config(anyhow!("Invalid metadata format")))
                }?;
                Ok(path)
            })
            .collect::<Result<Vec<_>>>()?;

        // Write manifest file to S3 instead of inserting to database
        if !paths.is_empty() {
            let s3_inner = self.config.s3_inner.as_ref().ok_or_else(|| {
                SinkError::Config(anyhow!("S3 configuration is required for S3 sink"))
            })?;
            {
                Self::write_manifest_to_s3(s3_inner, paths, &self.config.table).await?;
            }
            tracing::info!(
                "Manifest file written to S3 for sink id {} at epoch {}",
                self.sink_id,
                epoch
            );

View on GitHub (pinned to 6469eb736d)