{"record":{"id":"a65e9f67c3a352a5","repo":"apache/flink","slug":"unsupported-operation-s-for-row-kind-a65e9f","errorCode":null,"errorMessage":"Unsupported operation '%s' for row kind.","messagePattern":"Unsupported operation '(.+?)' for row kind\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonSerializationSchema.java","lineNumber":106,"sourceCode":"\n    @Override\n    public byte[] serialize(RowData rowData) {\n        try {\n            switch (rowData.getRowKind()) {\n                case INSERT:\n                case UPDATE_AFTER:\n                    genericRowData.setField(0, null);\n                    genericRowData.setField(1, rowData);\n                    genericRowData.setField(2, OP_INSERT);\n                    return jsonSerializer.serialize(genericRowData);\n                case UPDATE_BEFORE:\n                case DELETE:\n                    genericRowData.setField(0, rowData);\n                    genericRowData.setField(1, null);\n                    genericRowData.setField(2, OP_DELETE);\n                    return jsonSerializer.serialize(genericRowData);\n                default:\n                    throw new UnsupportedOperationException(\n                            format(\n                                    \"Unsupported operation '%s' for row kind.\",\n                                    rowData.getRowKind()));\n            }\n        } catch (Throwable t) {\n            throw new RuntimeException(format(\"Could not serialize row '%s'.\", rowData), t);\n        }\n    }\n\n    @Override\n    public boolean equals(Object o) {\n        if (this == o) {\n            return true;\n        }\n        if (o == null || getClass() != o.getClass()) {\n            return false;\n        }\n        OggJsonSerializationSchema that = (OggJsonSerializationSchema) o;","sourceCodeStart":88,"sourceCodeEnd":124,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonSerializationSchema.java#L88-L124","documentation":"OggJsonSerializationSchema.serialize switches on the input RowData's RowKind; only INSERT/UPDATE_AFTER (mapped to OP_INSERT) and UPDATE_BEFORE/DELETE (mapped to OP_DELETE) are handled. Any other RowKind falls to the default branch and throws UnsupportedOperationException, which is immediately wrapped by the catch into 'Could not serialize row'.","triggerScenarios":"A RowData with a RowKind outside the four CDC kinds reaching an ogg-json sink — custom operators, extended RowKind enums, or malformed test fixtures.","commonSituations":"Custom ProcessFunction tagging rows with non-standard kinds; Flink version drift adding new RowKind values; generic pipelines reused across CDC formats.","solutions":["Normalize RowKind to the four CDC values in a map immediately before the sink.","Search the job graph for setRowKind/fromShortValue calls that can produce exotic kinds and constrain them."],"exampleFix":"// before\ndata.setRowKind(RowKind.fromShortValue((short) 11));\nout.collect(data);\n// after\ndata.setRowKind(RowKind.DELETE);\nout.collect(data);","handlingStrategy":"type-guard","validationCode":null,"typeGuard":"private static final Set<RowKind> CDC_KINDS = EnumSet.of(\n        RowKind.INSERT, RowKind.UPDATE_BEFORE, RowKind.UPDATE_AFTER, RowKind.DELETE);\n\nstatic RowKind clampToCdc(RowKind k) {\n    return CDC_KINDS.contains(k) ? k : RowKind.INSERT;\n}","tryCatchPattern":null,"preventionTips":["Clamp kinds in the operator directly upstream of an ogg-json sink.","Ban custom RowKind codes in code review for pipelines ending in CDC formats."],"tags":["flink","ogg","json","serialization","rowkind"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}