{"record":{"id":"7f3385d07d907b56","repo":"risingwavelabs/risingwave","slug":"op-field-not-found-in-canal-json","errorCode":null,"errorMessage":"op field not found in canal json","messagePattern":"op field not found in canal json","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/parser/canal/simd_json_parser.rs","lineNumber":77,"sourceCode":"        mut payload: Vec<u8>,\n        mut writer: SourceStreamChunkRowWriter<'_>,\n    ) -> ConnectorResult<()> {\n        let mut event: BorrowedValue<'_> =\n            simd_json::to_borrowed_value(&mut payload[self.payload_start_idx..])\n                .context(\"failed to parse canal json payload\")?;\n\n        let is_ddl = event\n            .get(IS_DDL)\n            .and_then(|v| v.as_bool())\n            .context(\"field `isDdl` not found in canal json\")?;\n        if is_ddl {\n            bail!(\"received a DDL message, please set `canal.instance.filter.query.dml` to true.\");\n        }\n\n        let op = match event.get(OP).and_then(|v| v.as_str()) {\n            Some(CANAL_INSERT_EVENT | CANAL_UPDATE_EVENT) => ChangeEventOperation::Upsert,\n            Some(CANAL_DELETE_EVENT) => ChangeEventOperation::Delete,\n            _ => bail!(\"op field not found in canal json\"),\n        };\n\n        let events = event\n            .get_mut(DATA)\n            .and_then(|v| match v {\n                BorrowedValue::Array(array) => Some(array),\n                _ => None,\n            })\n            .context(\"field `data` is missing for creating event\")?;\n\n        let mut errors = Vec::new();\n        for event in events.drain(..) {\n            let accessor = JsonAccess::new_with_options(event, &JsonParseOptions::CANAL);\n            match apply_row_operation_on_stream_chunk_writer((op, accessor), &mut writer) {\n                Ok(_) => {}\n                Err(err) => errors.push(err),\n            }\n        }","sourceCodeStart":59,"sourceCodeEnd":95,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/parser/canal/simd_json_parser.rs#L59-L95","documentation":"After the DDL check, parse_inner reads the `op` field of the Canal JSON event. If it is missing or not one of INSERT/UPDATE/DELETE (Canal values INSERT, UPDATE, DELETE), the parser cannot map the event to a change operation and bails.","triggerScenarios":"Message payload lacks an `op` field, or `op` has an unexpected value (e.g. `CREATE`, `QUERY`, `TRUNCATE`) — anything not INSERT/UPDATE/DELETE.","commonSituations":"Non-Canal-format JSON pushed to the topic; Canal heartbeat or transaction-begin/end events (op=BEGIN/COMMIT) reaching the source; producer format drift or schema registry/message-format changes.","solutions":["Ensure the topic contains only Canal JSON DML data messages; enable heartbeat filtering (`canal.instance.filter.query.dml=true` and disable heartbeat/tx events in MQ routing)","Verify the message format is Canal JSON (has isDdl/op/data fields) by inspecting a raw Kafka message","Use a compatible source format (e.g. debezium json) if the upstream is not actually Canal"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"if (!msg || typeof msg.op !== 'string' || !['INSERT','UPDATE','DELETE'].includes(msg.op)) { drop(msg); }","typeGuard":"const isCanalDml = (m) => m && typeof m.op === 'string' && ['INSERT','UPDATE','DELETE'].includes(m.op);","tryCatchPattern":"if err.to_string().contains(\"op field not found\") { inspect_raw_kafka_message(topic, offset); } else { propagate }","preventionTips":["Verify Canal JSON format with kafka-console-consumer before wiring the source","Exclude heartbeat/transaction-control events from the MQ topic","Pin the producer message format; alert on format drift"],"tags":["canal","cdc","json","streaming-parser"],"backgroundTag":"unexpected-response-shape","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"}