{"record":{"id":"eb6e1a4ff512c7ec","repo":"apache/beam","slug":"recorddeltataskwriter-received-unsorted-input-a-record-s","errorCode":null,"errorMessage":"RecordDeltaTaskWriter received unsorted input: a record's sort key sorts below its predecessor's within the group.","messagePattern":"RecordDeltaTaskWriter received unsorted input: a record's sort key sorts below its predecessor's within the group\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java","lineNumber":171,"sourceCode":"                + \" is not a top-level column of schema: \"\n                + schema);\n      }\n      this.pkPos[i] = pos;\n    }\n    this.upsert = upsert;\n  }\n\n  /** Routes a record to the {@link PartitionDeltaWriter} responsible for its partition. */\n  abstract PartitionDeltaWriter route(Record row);\n\n  /**\n   * Buffers {@code row} into the current block, flushing the previous block first when {@code\n   * sortKey} starts a new primary key.\n   */\n  public void write(byte[] sortKey, Record row, ValueKind kind) {\n    // The collapse is only correct over sorted input, so a regressing key must not be accepted.\n    if (lastSortKey != null && Arrays.compareUnsigned(sortKey, lastSortKey) < 0) {\n      throw new IllegalStateException(\n          \"RecordDeltaTaskWriter received unsorted input: a record's sort key sorts below its \"\n              + \"predecessor's within the group.\");\n    }\n    lastSortKey = sortKey.clone();\n    if (blockKey != null && !CdcSortKey.samePk(blockKey, sortKey)) {\n      // we're encountering a new PK. flush the current one\n      flushBlock();\n    }\n    if (blockKey == null) {\n      blockKey = sortKey.clone();\n      firstRecord = row;\n      firstKind = kind;\n    }\n    if (kind == ValueKind.UPDATE_BEFORE || kind == ValueKind.DELETE) {\n      sawUbOrDelete = true;\n    }\n    latestRecord = row;\n    latestKind = kind;","sourceCodeStart":153,"sourceCodeEnd":189,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java#L153-L189","documentation":"The CDC collapse in RecordDeltaTaskWriter is only correct over input sorted by CdcSortKey. In write(), if the incoming sort key compares unsigned-less than the previously written key, the input has regressed within the group, and the writer throws IllegalStateException rather than produce incorrect collapsed output.","triggerScenarios":"Calling write(sortKey, row, kind) with a byte[] sortKey that sorts below the previous call's key — e.g. feeding an unsorted DoFn output, shuffling data incorrectly, or emitting records across partition boundaries out of order.","commonSituations":"Upstream sort pipeline dropped or was bypassed; a Beam reshuffle reordered elements; custom sources emit records without preserving the sort order the CDC sink requires.","solutions":["Ensure the PCollection feeding RecordDeltaTaskWriter is globally (or per-key) sorted by CdcSortKey before writing","Check that no stage between the sort and the writer drops the sort guarantee (insert an explicit sort or verify the comparator)","Log the offending sort key and its predecessor to identify which upstream stage emits out-of-order records"],"exampleFix":"// before\npipeline.apply(\"write\", new RecordDeltaTaskWriter(...)); // input unsorted\n// after\npipeline\n    .apply(\"sort\", Sort.reinstrumented(...) /* sort by CdcSortKey bytes */)\n    .apply(\"write\", new RecordDeltaTaskWriter(...));","handlingStrategy":"validation","validationCode":"byte[] prev = null;\nfor (Record r : input) {\n  byte[] k = CdcSortKey.of(r);\n  if (prev != null && Arrays.compareUnsigned(k, prev) < 0) {\n    throw new IllegalStateException(\"unsorted input upstream of writer\");\n  }\n  prev = k;\n}","typeGuard":null,"tryCatchPattern":"try {\n  writer.write(sortKey, row, kind);\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"unsorted input\")) {\n    // route to unsorted-input dead-letter / re-sort stage\n  } else throw e;\n}","preventionTips":["Always place an explicit sort (by CdcSortKey bytes) immediately before the writer","Avoid reshuffles between the sort and the writer that don't preserve order","Add an order assertion in tests for the sort stage"],"tags":["iceberg","cdc","input-ordering"],"backgroundTag":"invalid-argument-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"}