{"record":{"id":"be65c28fbca87f1a","repo":"risingwavelabs/risingwave","slug":"avro-error-0","errorCode":null,"errorMessage":"Avro error: {0}","messagePattern":"Avro error: (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1106,"sourceCode":"#[derive(Error, Debug)]\npub enum SinkError {\n    #[error(\"Kafka error: {0}\")]\n    Kafka(#[from] rdkafka::error::KafkaError),\n    #[error(\"Kinesis error: {0}\")]\n    Kinesis(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Remote sink error: {0}\")]\n    Remote(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Encode error: {0}\")]\n    Encode(String),\n    #[error(\"Avro error: {0}\")]\n    Avro(#[from] apache_avro::Error),\n    #[error(\"Iceberg error: {0}\")]\n    Iceberg(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"config error: {0}\")]\n    Config(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"coordinator error: {0}\")]\n    Coordinator(\n        #[source]\n        #[backtrace]\n        anyhow::Error,","sourceCodeStart":1088,"sourceCodeEnd":1124,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1088-L1124","documentation":"This is a variant of `SinkError` in RisingWave's connector crate, defined via thiserror. It wraps any `apache_avro::Error` via `#[from]`, so any failure inside the Avro serialization/deserialization machinery used by the Avro sink (or schema-registry based sinks) is surfaced as \"Avro error: {0}\". The original error's source and backtrace are preserved for diagnostics.","triggerScenarios":"Constructing or writing to the Avro sink: encoding a Row into Avro bytes, schema (de)registration with the schema registry, parsing the user-supplied Avro schema, or resolving/datum-conversion failures that return `apache_avro::Error`.","commonSituations":"Invalid user-provided Avro schema JSON in the sink's `type`/schema option; a RisingWave column type that cannot map to the declared Avro field type (type mismatch during datum conversion); schema-registry incompatibility or schema evolution conflicts (new schema not compatible with registered versions); corrupted or incompatible apache-avro crate behavior on unusual types (e.g. Decimal/Date precision).","solutions":["Read the wrapped message (the `{0}` payload) — it names the exact apache-avro failure; fix the schema or data per that message.","Validate the Avro schema passed to the sink: load it with `apache_avro::Schema::parse_str` locally to confirm it is valid JSON/IDL.","Check that RisingWave column types match the Avro schema field types (e.g. timestamp precision, decimal scale) and adjust the schema or `encode` options.","If the error arises during schema-registry interaction, verify registry connectivity and that the schema is compatible with previously registered versions under the same subject."],"exampleFix":"-- before\nCREATE SINK s FROM mv WITH (\n  connector = 'kafka',\n  type = '<invalid avro json>',\n  ...\n);\n-- after\nCREATE SINK s FROM mv WITH (\n  connector = 'kafka',\n  format = 'debezium_avro',\n  type = '{\"type\":\"record\",\"name\":\"envelope\",\"fields\":[...]}',\n  ...\n);","handlingStrategy":"try-catch","validationCode":"use apache_avro::Schema;\nfn validate_avro_schema(schema_json: &str) -> Result<(), String> {\n    Schema::parse_str(schema_json)\n        .map(|_| ())\n        .map_err(|e| format!(\"invalid avro schema: {e}\"))\n}","typeGuard":"fn as_avro_error(err: &SinkError) -> Option<&apache_avro::Error> {\n    if let SinkError::Avro(e) = err { Some(e) } else { None }\n}","tryCatchPattern":"match sink.write(batch).await {\n    Err(SinkError::Avro(e)) => {\n        log::error!(\"avro serialization failed: {e}\");\n        // inspect schema/types, then retry or poison the sink\n    }\n    Err(e) => return Err(e.into()),\n    Ok(_) => {}\n}","preventionTips":["Validate the Avro schema with apache_avro::Schema::parse_str before creating the sink.","Keep RisingWave column types aligned with Avro schema field types (decimal scale, timestamp precision).","When evolving schemas, ensure backward compatibility with versions registered in the schema registry.","Pin and test against the apache-avro crate version RisingWave uses when doing local repro."],"tags":["avro","serialization","sink","risingwave"],"backgroundTag":"schema-validation-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}