risingwavelabs/risingwave · error · SinkError::Config

topic field `{}` not found

Error message

topic field `{}`  not found

What it means

`get_topic_field_index_path` cannot find the column named by the `topic_field` option in the sink's schema, so it cannot resolve an index path into the message. This error means the configured topic field does not exist on the data being sunk.

Source

Thrown at src/connector/src/sink/mqtt.rs:474

                        st.iter().enumerate().find(|(_, (s, _))| *s == field).map(
                            |(pos, (_, dt))| {
                                path.push(pos);
                                dt
                            },
                        )
                    }
                    _ => None,
                })
            });

    match dt {
        Some(DataType::Varchar) => Ok(path),
        Some(dt) => Err(SinkError::Config(anyhow!(
            "topic field `{}` must be of type string but got {:?}",
            topic_field,
            dt
        ))),
        None => Err(SinkError::Config(anyhow!(
            "topic field `{}`  not found",
            topic_field
        ))),
    }
}

#[cfg(test)]
mod test {
    use risingwave_common::array::{DataChunk, DataChunkTestExt, RowRef};
    use risingwave_common::catalog::{Field, Schema};
    use risingwave_common::types::{DataType, StructType};

    use super::{get_topic_field_index_path, get_topic_from_index_path};

    #[test]
    fn test_single_field_extraction() {
        let schema = Schema::new(vec![Field::with_name(DataType::Varchar, "topic")]);
        let path = get_topic_field_index_path(&schema, "topic").unwrap();

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the sink schema and correct the `topic_field` name to match an existing column exactly
2. If the column is not in the sinked stream, add it to the SELECT feeding the sink
3. Re-run `CREATE SINK` after fixing the option

Example fix

// before
CREATE SINK s FROM t WITH (connector='mqtt', topic_field='devce_id');
// after
CREATE SINK s FROM t WITH (connector='mqtt', topic_field='device_id');

// ensure topic_field exists in the sink schema
assert!(
    schema.fields().iter().any(|f| f.name.as_str() == topic_field),
    "topic_field '{}' not found in schema {:?}", topic_field, schema.fields()
);

fn topic_field_exists(schema: &Schema, name: &str) -> bool {
    schema.fields().iter().any(|f| f.name.as_str() == name)
}

When it happens

Trigger: Creating/validating an MQTT sink where the `topic_field` option names a column absent from the sink's schema (typo, wrong case, or column not included in the sink's SELECT).

Common situations: Typo in the `topic_field` value; user assumes a column exists in the sinked data but the SELECT/project excludes it; upstream rename of the column.

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

• topic field `{}` must be of type string but got {:?}
• sink type unsupported: {}
• sink format unsupported: {}
• sink encode unsupported: {}
• unsupported {} as sink key encode

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