apache/iceberg · error · IllegalStateException

Incoming records violate the writer assumption that records

Error message

Incoming records violate the writer assumption that records are clustered by spec and by partition within each spec. Either cluster the incoming records or switch to fanout writers.
Encountered records that belong to already closed files:
partition '<path>' in spec <spec>

What it means

ClusteredWriter.write also enforces clustering by partition within a spec: once a partition's writer is closed, records for that same partition must not appear again. When a record's partition key matches an already-completed partition, it throws IllegalStateException naming the partition path and spec.

Source

Thrown at core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java:96

      StructType partitionType = spec.partitionType();

      this.currentSpec = spec;
      this.partitionComparator = Comparators.forType(partitionType);
      this.completedPartitions = StructLikeSet.create(partitionType);
      // copy the partition key as the key object may be reused
      this.currentPartition = StructLikeUtil.copy(partition);
      this.currentWriter = newWriter(currentSpec, currentPartition);

    } else if (partition != currentPartition
        && partitionComparator.compare(partition, currentPartition) != 0) {
      closeCurrentWriter();
      completedPartitions.add(currentPartition);

      if (completedPartitions.contains(partition)) {
        String errorCtx =
            String.format("partition '%s' in spec %s", spec.partitionToPath(partition), spec);
        throw new IllegalStateException(NOT_CLUSTERED_ROWS_ERROR_MSG_TEMPLATE + errorCtx);
      }

      // copy the partition key as the key object may be reused
      this.currentPartition = StructLikeUtil.copy(partition);
      this.currentWriter = newWriter(currentSpec, currentPartition);
    }

    currentWriter.write(row);
  }

  @Override
  public void close() throws IOException {
    if (!closed) {
      closeCurrentWriter();
      this.closed = true;
    }
  }

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Fully sort/repartition records by partition within each spec before writing.
2. Switch to a fanout writer that keeps writers open for all partitions.
3. Batch records grouped by partition and flush each group to a dedicated ClusteredWriter instance.

Example fix

// before
records.forEach(clusteredWriter::write); // interleaved partitions
// after
records.stream()
    .sorted(Comparator.comparing(r -> partitionKey(r)))
    .forEach(clusteredWriter::write);

// ensure records are sorted by partition within each spec
records = records.sorted(Comparator.comparing((T r) -> specIdOf(r)).thenComparing(r -> partitionKey(r)));

try {
  writer.write(record);
} catch (IllegalStateException e) {
  // partition reappeared after close: sort input or use fanout writer
}

When it happens

Trigger: Writing records to a clustered writer where partition P, then a different partition, then partition P again occur within the same spec — the writer for P was already closed so it cannot be reopened.

Common situations: Input sorted by spec but not fully by partition (e.g., partial sort, multiple upstream tasks each emitting overlapping partitions); key-reuse bugs where the comparator sees the partition as equal but subsequent writes revisit it.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Incoming records violate the writer assumption that records
• Failed to close equality delta writer
• Already closed files for partition:
• Cannot bucket by type:
• Identity transform is not supported

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/310d60175aabd2c2. Report an issue: GitHub.