{"record":{"id":"6566d8d1d97dcce3","repo":"risingwavelabs/risingwave","slug":"encode-error-0","errorCode":null,"errorMessage":"Encode error: {0}","messagePattern":"Encode error: (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1104,"sourceCode":"pub type Result<T> = std::result::Result<T, SinkError>;\n\n#[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]","sourceCodeStart":1086,"sourceCodeEnd":1122,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1086-L1122","documentation":"Variant of `SinkError` produced when the sink's encoder fails, typically because incoming row data cannot be encoded into the target format (JSON/Avro/Protobuf/Debezium) according to the sink schema — e.g. a payload that doesn't validate against the registered schema.","triggerScenarios":"Encoding a stream chunk during sink write when the row violates the target format: a field type mismatch vs the Avro schema, missing NOT NULL-compatible field, invalid protobuf payload, or a malformed value the encoder cannot serialize.","commonSituations":"Upstream schema evolved but the sink's registered Avro/JSON schema (schema registry) is stale; decimal/nullable handling differences; sink created before an ALTER added columns the encoder can't map.","solutions":["Read the inner message to see which field/value failed encoding and fix the upstream data or the column type","Recreate or alter the sink so its schema matches the current source (e.g. re-register the Avro schema with schema registry)","Cast or sanitize problematic columns upstream (`SELECT CAST(...)` in the sink's query) so they fit the target format"],"exampleFix":"-- before\nCREATE SINK s FROM mv INTO kafka WITH (connector='kafka', format='avro', ...);\n-- fails on incompatible decimal column\n\n-- after\nCREATE SINK s FROM (SELECT id, CAST(amount AS DOUBLE) AS amount FROM mv) INTO kafka\n  WITH (connector='kafka', format='avro', ...);","handlingStrategy":"validation","validationCode":"// ensure every column in the sink query matches the target schema before CREATE SINK\n-- compare: SELECT column_name, data_type FROM rw_columns WHERE relation = 'mv';\n-- against the registered Avro/JSON/Protobuf schema","typeGuard":null,"tryCatchPattern":"match err {\n    SinkError::Encode(msg) => {\n        log::error!(\"sink encode failed: {msg}\"); // inspect field/value reported, fix data or schema\n    }\n    e => return Err(e.into()),\n}","preventionTips":["After ALTERs on sources, re-check sink schema compatibility (schema registry) before resuming","Cast columns explicitly in the sink query to types the target format supports","Use schema evolution features/versions supported by your connector instead of ad-hoc schema edits"],"tags":["sink","encoding","avro","schema"],"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-14T11:17:12.474Z"}