apache/iceberg · error · IllegalStateException

Already closed files for partition:

Error message

Already closed files for partition: 

What it means

PartitionedWriter.write() enforces that all rows for a given partition are supplied contiguously, because files for a partition are closed as soon as the partition key changes. Writing a row for a partition whose files were already closed throws this IllegalStateException so incorrectly grouped input fails loudly instead of corrupting data.

Source

Thrown at core/src/main/java/org/apache/iceberg/io/PartitionedWriter.java:85

  protected abstract PartitionKey partition(T row);

  @Override
  public void write(T row) throws IOException {
    PartitionKey key = partition(row);

    if (!key.equals(currentKey)) {
      if (currentKey != null) {
        // if the key is null, there was no previous current key and current writer.
        currentWriter.close();
        completedPartitions.add(currentKey);
      }

      if (completedPartitions.contains(key)) {
        // if rows are not correctly grouped, detect and fail the write
        PartitionKey existingKey = Iterables.find(completedPartitions, key::equals, null);
        LOG.warn("Duplicate key: {} == {}", existingKey, key);
        throw new IllegalStateException("Already closed files for partition: " + key.toPath());
      }

      currentKey = key.copy();
      currentWriter = new RollingFileWriter(currentKey);
    }

    currentWriter.write(row);
  }

  @Override
  public void close() throws IOException {
    if (currentWriter != null) {
      currentWriter.close();
    }
  }
}

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Re-partition/sort the incoming data by the table's partition key before writing (e.g. set write.distribution-mode=hash so the framework adds a keyBy)
2. In Flink, ensure distributeDataStream applies PartitionKeySelector before the writer operator
3. Check for operator chaining/parallelism changes (rebalancing/rescaling) that break ordering between partitioning and the writer
4. Presort batch records by partition key before append
5. Use unpartitioned writes or a fanout writer if contiguity cannot be guaranteed

Example fix

// before (Spark)
df.writeTo(table).append()  // distribution-mode none, rows unsorted
// after
spark.conf.set("write.distribution-mode", "hash")
df.writeTo(table).append()

// verify rows are grouped by partition key before writing
Object prev = null;
Set<Object> seen = new HashSet<>();
for (Row r : rows) {
  Object k = partitionKeyValue(r);
  if (prev != null && !prev.equals(k) && seen.contains(k)) {
    throw new IllegalArgumentException("rows not grouped by partition key");
  }
  seen.add(k); prev = k;
}

try {
  writer.write(row);
} catch (IllegalStateException e) {
  // restart task with correct hash distribution; do not retry in-place
}

When it happens

Trigger: Calling write() with a PartitionKey that is contained in completedPartitions — i.e. rows arrive as A, B, A: the second A hits a closed partition. Typical causes are an upstream keyBy/partitioning that does not group by the Iceberg partition key, or unsorted non-hash-distributed input.

Common situations: Flink/Spark write pipelines missing the required keyBy/distribute stage before the Iceberg writer; changing write.distribution-mode without re-partitioning; feeding arbitrarily ordered batches (e.g. from an unordered source) into a partitioned writer.

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

• Failed to close equality delta writer
• Incoming records violate the writer assumption that records
• Incoming records violate the writer assumption that records
• Altering partition keys is not supported yet.
• Failed to encode partition

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