risingwavelabs/risingwave · error · SinkError::Config

(dynamic: UTF-8 conversion error from metadata bytes)

Error message

(dynamic: UTF-8 conversion error from metadata bytes)

What it means

During `commit_data` with S3 manifests, each sink metadata entry must be a `Serialized` metadata variant whose bytes decode as UTF-8 (the file path written to S3). This message is the dynamic UTF-8 conversion error text produced when the serialized metadata bytes are not valid UTF-8.

Solutions

1. Inspect the upstream data/metadata generation to find where non-UTF-8 bytes originate
2. Ensure the metadata column contains valid UTF-8 path strings
3. Replay the affected epoch or skip corrupted metadata entries

// validate metadata bytes are UTF-8 before emitting them as sink metadata
std::str::from_utf8(&bytes).map_err(|_| "non-UTF-8 sink metadata")?;

fn valid_path_meta(m: &Option<SinkMetadataEnum>) -> Option<String> {
    match m.as_ref()? {
        Metadata::Serialized(SerializedMetadata { metadata }) =>
            String::from_utf8(metadata.clone()).ok(),
        _ => None,
    }
}

When it happens

Trigger: A `sink_metadata::Metadata::Serialized` payload contains bytes that fail `String::from_utf8` — i.e., something upstream wrote non-UTF-8 data as the metadata path.

Common situations: Custom/modified sink code putting raw binary into Serialized metadata; corrupted data in the stream's metadata column; version mismatch between producer and consumer of the metadata format.

Understand the failure class

Background: "Invalid ... format", "must be in format X", "does not look like a ..." — invalid argument format errors across CLI tools and libraries — this error's family across 17 libraries.

Related errors

• Invalid metadata format
• Can't create iceberg sink write result from empty data!
• (dynamic: channel send error on shutdown signal)
• iceberg sink metadata should be an object
• intermediate.table.name is required for append-only sink

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

Appendix: source

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

            }
        };
        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,

View on GitHub (pinned to 6469eb736d)