risingwavelabs/risingwave · error · SinkError::Config

`write_batch_size` must be greater than 0

Error message

`write_batch_size` must be greater than 0

What it means

Numeric option validation in TurbopufferConfig::from_btreemap: the `write_batch_size` option was set to 0 (or otherwise non-positive), which would make sink batches empty and stall writing. The guard enforces a strictly positive batch size when the sink config is built.

Solutions

  1. Set write_batch_size to a positive integer (e.g. 500)
  2. Remove the write_batch_size option to use the default
  3. Validate the value before issuing CREATE SINK

Example fix

// before
WITH (connector='turbopuffer', write_batch_size = '0')
// after
WITH (connector='turbopuffer', write_batch_size = '500')
Defensive patterns

Strategy: validation

Validate before calling

if (props.write_batch_size !== undefined && Number(props.write_batch_size) <= 0) throw new Error('write_batch_size must be > 0')

Type guard

const validBatch = (n) => Number.isInteger(n) && n > 0;

Try / catch

catch (SinkError::Config(e)) if e.contains('write_batch_size') { fix option and retry DDL }

Prevention

When it happens

Trigger: CREATE SINK with connector='turbopuffer' and WITH option write_batch_size = '0'.

Common situations: Copy-pasted config with 0 placeholder; computed batch size evaluating to 0; misunderstanding of default behavior (omit the option to use default).

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

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

Appendix: source

Thrown at src/connector/src/sink/turbopuffer.rs:102

    pub unknown_fields: std::collections::HashMap<String, String>,
}

crate::impl_sink_unknown_fields!(TurbopufferConfig);

impl EnforceSecret for TurbopufferConfig {
    const ENFORCE_SECRET_PROPERTIES: phf::Set<&'static str> = phf::phf_set! {
        "api_key",
    };
}

impl TurbopufferConfig {
    fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {
        let config = serde_json::from_value::<TurbopufferConfig>(
            serde_json::to_value(values).expect("serialize sink properties"),
        )
        .map_err(|e| SinkError::Config(anyhow!(e)))?;
        if config.write_batch_size == 0 {
            return Err(SinkError::Config(anyhow!(
                "`write_batch_size` must be greater than 0"
            )));
        }
        if config.max_linger_second == 0 {
            return Err(SinkError::Config(anyhow!(
                "`max_linger_second` must be greater than 0"
            )));
        }
        if config.num_shards == Some(0) {
            return Err(SinkError::Config(anyhow!(
                "`num_shards` must be greater than 0"
            )));
        }
        Ok(config)
    }
}

#[derive(Clone, Debug)]

View on GitHub (pinned to 6469eb736d)