apache/beam · error · IllegalStateException

RecordDeltaTaskWriter received unsorted input: a record's…

Error message

RecordDeltaTaskWriter received unsorted input: a record's sort key sorts below its predecessor's within the group.

What it means

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.

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

Example fix

// before
pipeline.apply("write", new RecordDeltaTaskWriter(...)); // input unsorted
// after
pipeline
    .apply("sort", Sort.reinstrumented(...) /* sort by CdcSortKey bytes */)
    .apply("write", new RecordDeltaTaskWriter(...));

Defensive patterns

Strategy: validation

Validate before calling

byte[] prev = null;
for (Record r : input) {
  byte[] k = CdcSortKey.of(r);
  if (prev != null && Arrays.compareUnsigned(k, prev) < 0) {
    throw new IllegalStateException("unsorted input upstream of writer");
  }
  prev = k;
}

Try / catch

try {
  writer.write(sortKey, row, kind);
} catch (IllegalStateException e) {
  if (e.getMessage().contains("unsorted input")) {
    // route to unsorted-input dead-letter / re-sort stage
  } else throw e;
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/eb6e1a4ff512c7ec. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java:171

                + " is not a top-level column of schema: "
                + schema);
      }
      this.pkPos[i] = pos;
    }
    this.upsert = upsert;
  }

  /** Routes a record to the {@link PartitionDeltaWriter} responsible for its partition. */
  abstract PartitionDeltaWriter route(Record row);

  /**
   * Buffers {@code row} into the current block, flushing the previous block first when {@code
   * sortKey} starts a new primary key.
   */
  public void write(byte[] sortKey, Record row, ValueKind kind) {
    // The collapse is only correct over sorted input, so a regressing key must not be accepted.
    if (lastSortKey != null && Arrays.compareUnsigned(sortKey, lastSortKey) < 0) {
      throw new IllegalStateException(
          "RecordDeltaTaskWriter received unsorted input: a record's sort key sorts below its "
              + "predecessor's within the group.");
    }
    lastSortKey = sortKey.clone();
    if (blockKey != null && !CdcSortKey.samePk(blockKey, sortKey)) {
      // we're encountering a new PK. flush the current one
      flushBlock();
    }
    if (blockKey == null) {
      blockKey = sortKey.clone();
      firstRecord = row;
      firstKind = kind;
    }
    if (kind == ValueKind.UPDATE_BEFORE || kind == ValueKind.DELETE) {
      sawUbOrDelete = true;
    }
    latestRecord = row;
    latestKind = kind;

View on GitHub (pinned to 12126d8942)