{"record":{"id":"ac6ed286138a8cca","repo":"apache/iceberg","slug":"already-closed-files-for-partition","errorCode":null,"errorMessage":"Already closed files for partition: ","messagePattern":"Already closed files for partition: ","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"core/src/main/java/org/apache/iceberg/io/PartitionedWriter.java","lineNumber":85,"sourceCode":"  protected abstract PartitionKey partition(T row);\n\n  @Override\n  public void write(T row) throws IOException {\n    PartitionKey key = partition(row);\n\n    if (!key.equals(currentKey)) {\n      if (currentKey != null) {\n        // if the key is null, there was no previous current key and current writer.\n        currentWriter.close();\n        completedPartitions.add(currentKey);\n      }\n\n      if (completedPartitions.contains(key)) {\n        // if rows are not correctly grouped, detect and fail the write\n        PartitionKey existingKey = Iterables.find(completedPartitions, key::equals, null);\n        LOG.warn(\"Duplicate key: {} == {}\", existingKey, key);\n        throw new IllegalStateException(\"Already closed files for partition: \" + key.toPath());\n      }\n\n      currentKey = key.copy();\n      currentWriter = new RollingFileWriter(currentKey);\n    }\n\n    currentWriter.write(row);\n  }\n\n  @Override\n  public void close() throws IOException {\n    if (currentWriter != null) {\n      currentWriter.close();\n    }\n  }\n}\n","sourceCodeStart":67,"sourceCodeEnd":101,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/core/src/main/java/org/apache/iceberg/io/PartitionedWriter.java#L67-L101","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["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)","In Flink, ensure distributeDataStream applies PartitionKeySelector before the writer operator","Check for operator chaining/parallelism changes (rebalancing/rescaling) that break ordering between partitioning and the writer","Presort batch records by partition key before append","Use unpartitioned writes or a fanout writer if contiguity cannot be guaranteed"],"exampleFix":"// before (Spark)\ndf.writeTo(table).append()  // distribution-mode none, rows unsorted\n// after\nspark.conf.set(\"write.distribution-mode\", \"hash\")\ndf.writeTo(table).append()","handlingStrategy":"validation","validationCode":"// verify rows are grouped by partition key before writing\nObject prev = null;\nSet<Object> seen = new HashSet<>();\nfor (Row r : rows) {\n  Object k = partitionKeyValue(r);\n  if (prev != null && !prev.equals(k) && seen.contains(k)) {\n    throw new IllegalArgumentException(\"rows not grouped by partition key\");\n  }\n  seen.add(k); prev = k;\n}","typeGuard":null,"tryCatchPattern":"try {\n  writer.write(row);\n} catch (IllegalStateException e) {\n  // restart task with correct hash distribution; do not retry in-place\n}","preventionTips":["Always set write.distribution-mode=hash for partitioned writes unless input is known sorted","Keep partitioning operator chained with the writer (no rebalance/rescale between them)","Presort batch inputs by partition key before append","Prefer unpartitioned or fanout writers when order cannot be guaranteed"],"tags":["writer","partitioning","flink","spark","data-ordering"],"backgroundTag":"invalid-state-transition","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}