apache/flink · error · UnsupportedOperationException

Unsupported operation

Error message

Unsupported operation '%s' for row kind.

What it means

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'.

Solutions

1. Normalize RowKind to the four CDC values in a map immediately before the sink.
2. Search the job graph for setRowKind/fromShortValue calls that can produce exotic kinds and constrain them.

Example fix

// before
data.setRowKind(RowKind.fromShortValue((short) 11));
out.collect(data);
// after
data.setRowKind(RowKind.DELETE);
out.collect(data);

private static final Set<RowKind> CDC_KINDS = EnumSet.of(
        RowKind.INSERT, RowKind.UPDATE_BEFORE, RowKind.UPDATE_AFTER, RowKind.DELETE);

static RowKind clampToCdc(RowKind k) {
    return CDC_KINDS.contains(k) ? k : RowKind.INSERT;
}

When it happens

Trigger: A RowData with a RowKind outside the four CDC kinds reaching an ogg-json sink — custom operators, extended RowKind enums, or malformed test fixtures.

Common situations: Custom ProcessFunction tagging rows with non-standard kinds; Flink version drift adding new RowKind values; generic pipelines reused across CDC formats.

Related errors

• Could not serialize row
• Unsupported operation
• Unsupported operation
• Corrupt Ogg JSON message
• Could not serialize row

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/a65e9f67c3a352a5. Report an issue: GitHub.

Appendix: source

Thrown at flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonSerializationSchema.java:106

    @Override
    public byte[] serialize(RowData rowData) {
        try {
            switch (rowData.getRowKind()) {
                case INSERT:
                case UPDATE_AFTER:
                    genericRowData.setField(0, null);
                    genericRowData.setField(1, rowData);
                    genericRowData.setField(2, OP_INSERT);
                    return jsonSerializer.serialize(genericRowData);
                case UPDATE_BEFORE:
                case DELETE:
                    genericRowData.setField(0, rowData);
                    genericRowData.setField(1, null);
                    genericRowData.setField(2, OP_DELETE);
                    return jsonSerializer.serialize(genericRowData);
                default:
                    throw new UnsupportedOperationException(
                            format(
                                    "Unsupported operation '%s' for row kind.",
                                    rowData.getRowKind()));
            }
        } catch (Throwable t) {
            throw new RuntimeException(format("Could not serialize row '%s'.", rowData), t);
        }
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        OggJsonSerializationSchema that = (OggJsonSerializationSchema) o;

View on GitHub (pinned to 2f3c205e92)