The key is {@code [pkLen:4][pkBytes][seq ^ Long.MIN_VALUE:8][kindRank:1]}, big-endian.\n */\nfinal class CdcSortKey {\n\n private CdcSortKey() {}\n\n /** Ranks change kinds so before-images sort before after-images at an equal {@code seq}. */\n public static byte kindRank(ValueKind kind) {\n switch (kind) {\n case UPDATE_BEFORE:\n return 0;\n case DELETE:\n return 1;\n case UPDATE_AFTER:\n return 2;\n case INSERT:\n return 3;\n default:\n throw new IllegalArgumentException(\"Unknown ValueKind: \" + kind);\n }\n }\n\n /**\n * Encodes the deterministic, byte-comparable sort key {@code [pkLen:4][pkBytes][seq ^\n * Long.MIN_VALUE:8][kindRank:1]} for one CDC record.\n *\n *
SortValues compares unsigned lexicographic byte order. The length prefix is needed to\n * accurately compare two primary keys of varying byte-lengths. Flipping the sequence number's\n * sign bit makes unsigned byte order match signed numeric order. kindRank breaks equal-seq ties.\n */\n public static byte[] encode(byte[] pkBytes, long seq, ValueKind kind) {\n return ByteBuffer.allocate(4 + pkBytes.length + 9)\n .putInt(pkBytes.length)\n .put(pkBytes)\n .putLong(seq ^ Long.MIN_VALUE)\n .put(kindRank(kind))\n .array();","sourceCodeStart":29,"sourceCodeEnd":65,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKey.java#L29-L65","documentation":"CdcSortKey.kindRank maps a CDC ValueKind (DELETE, UPDATE_BEFORE, UPDATE_AFTER, INSERT) to a one-byte rank used in the deterministic sort key. A ValueKind outside these four throws IllegalArgumentException, guaranteeing the sort encoding stays byte-comparable and deterministic for all known kinds.","triggerScenarios":"Calling kindRank (directly or via encode) with a ValueKind not in {DELETE, UPDATE_BEFORE, UPDATE_AFTER, INSERT} — e.g. a custom kind or a new kind added by a newer CDC record model.","commonSituations":"Upgrading the record model and adding a ValueKind without updating kindRank; constructing CdcUpdate kinds manually with an out-of-enum value; deserializing records produced by a different pipeline version with extended kinds.","solutions":["Only use the four ValueKinds supported by CdcSortKey","Add a case for the new ValueKind with a defined rank before encoding sort keys","Validate kinds at record-construction time to fail early","Ensure producer and consumer pipeline versions agree on the kind enum"],"exampleFix":"// before\nCdcUpdate update = new CdcUpdate(customKind, pk, seq, row);\nbyte[] key = CdcSortKey.encode(update); // throws\n// after\nif (kind != ValueKind.DELETE && kind != ValueKind.UPDATE_BEFORE\n && kind != ValueKind.UPDATE_AFTER && kind != ValueKind.INSERT) {\n throw new IllegalArgumentException(\"Kind not supported for sort key: \" + kind);\n}\nbyte[] key = CdcSortKey.encode(update);","handlingStrategy":"validation","validationCode":"if (kind != ValueKind.DELETE && kind != ValueKind.UPDATE_BEFORE\n && kind != ValueKind.UPDATE_AFTER && kind != ValueKind.INSERT) {\n throw new IllegalArgumentException(\"ValueKind not encodable in sort key: \" + kind);\n}","typeGuard":"static boolean isEncodableKind(ValueKind kind) {\n return kind == ValueKind.DELETE\n || kind == ValueKind.UPDATE_BEFORE\n || kind == ValueKind.UPDATE_AFTER\n || kind == ValueKind.INSERT;\n}","tryCatchPattern":"try {\n byte[] key = CdcSortKey.encode(update);\n} catch (IllegalArgumentException e) {\n LOG.error(\"Unencodable ValueKind for sort key: {}\", e.getMessage());\n throw e;\n}","preventionTips":["Only construct CDC records with the four supported ValueKinds","Update kindRank whenever the ValueKind enum gains a member","Keep producer and consumer pipeline versions aligned on the kind enum"],"tags":["iceberg","cdc","sort-key","enum"],"backgroundTag":"invalid-enum-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}