{"record":{"id":"fce06e9b42c702d7","repo":"apache/flink","slug":"unknown-op-value-s-the-debezium-json-message","errorCode":null,"errorMessage":"Unknown \"op\" value \"%s\". The Debezium JSON message is '%s'","messagePattern":"Unknown \"op\" value \"(.+?)\"\\. The Debezium JSON message is '(.+?)'","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java","lineNumber":180,"sourceCode":"            } else if (OP_UPDATE.equals(op)) {\n                if (before == null) {\n                    throw new IllegalStateException(\n                            String.format(REPLICA_IDENTITY_EXCEPTION, \"UPDATE\"));\n                }\n                before.setRowKind(RowKind.UPDATE_BEFORE);\n                after.setRowKind(RowKind.UPDATE_AFTER);\n                genericRowDataList.add(handleRow(row, before));\n                genericRowDataList.add(handleRow(row, after));\n            } else if (OP_DELETE.equals(op)) {\n                if (before == null) {\n                    throw new IllegalStateException(\n                            String.format(REPLICA_IDENTITY_EXCEPTION, \"DELETE\"));\n                }\n                before.setRowKind(RowKind.DELETE);\n                genericRowDataList.add(handleRow(row, before));\n            } else {\n                if (!ignoreParseErrors) {\n                    throw new IOException(\n                            format(\n                                    \"Unknown \\\"op\\\" value \\\"%s\\\". The Debezium JSON message is '%s'\",\n                                    op, new String(message)));\n                }\n                if (LOG.isDebugEnabled()) {\n                    LOG.debug(\n                            \"Unknown \\\"op\\\" value '{}'. The Debezium JSON message is '{}'.\",\n                            op,\n                            new String(message));\n                }\n            }\n        } catch (Throwable t) {\n            // a big try catch to protect the processing.\n            if (!ignoreParseErrors) {\n                throw new IOException(\n                        format(\"Corrupt Debezium JSON message '%s'.\", new String(message)), t);\n            }\n            if (LOG.isDebugEnabled()) {","sourceCodeStart":162,"sourceCodeEnd":198,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java#L162-L198","documentation":"Thrown by DebeziumJsonDeserializationSchema when a Debezium message's \"op\" field is not one of 'c' (create), 'r' (read/snapshot), 'u' (update), 'd' (delete) — e.g. truncate ('t') or no-op events. With 'json.ignore-parse-errors' disabled (default) this IOException fails the job; with it enabled the message is logged at DEBUG and skipped.","triggerScenarios":"A Debezium topic containing 't' (truncate), 'n' (heartbeat/signal op variants), or vendor/connector-specific op values, consumed via format 'debezium-json' with default options.","commonSituations":"Debezium TRUNCATE events enabled via snapshot.mode/endpoint options (Postgres/MySQL truncate handling); mixed topics with signal or heartbeat events; newer Debezium versions emitting additional op kinds; mislabeled topics carrying non-Debezium JSON.","solutions":["Set 'json.ignore-parse-errors' = true to skip unsupported op types when losing truncate/heartbeat events is acceptable","Disable TRUNCATE emission at the Debezium connector if those events are not needed (e.g. timestamper/skipped ddl options depending on connector)","Route only data-change events to the consumed topic via connector's topic routing/filter (SMT) configuration","Verify the payload is genuinely Debezium JSON ('op' at payload level) and not schema-wrapped twice"],"exampleFix":"// before\nWITH ('connector'='kafka', 'format'='debezium-json')\n\n// after\nWITH ('connector'='kafka', 'format'='debezium-json',\n  'json.ignore-parse-errors'='true')","handlingStrategy":"fallback","validationCode":"String op = payload.get(\"op\").asText();\nif (!Set.of(\"c\",\"r\",\"u\",\"d\").contains(op)) { /* skip / route elsewhere */ }","typeGuard":"static boolean isHandledDebeziumOp(String op) {\n    return op != null && op.length() == 1 && \"crud\".indexOf(op) >= 0; // c,u,d plus r snapshot\n}","tryCatchPattern":"catch (IOException e) on 'Unknown \"op\" value' — enable ignore-parse-errors for benign ops (t/heartbeat) or filter them at the connector; no point retrying.","preventionTips":["Disable truncate/heartbeat emission at the Debezium connector when not needed","Know your connector's op vocabulary before consuming","Monitor skipped events when ignore-parse-errors is enabled"],"tags":["debezium","cdc","unknown-op","option","runtime"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}