{"record":{"id":"cbc9a98f7e18db7d","repo":"apache/iceberg","slug":"internal-algorithm-error-exhausted-subtasks-with-cbc9a9","errorCode":null,"errorMessage":"Internal algorithm error: exhausted subtasks with unassigned keys left","messagePattern":"Internal algorithm error: exhausted subtasks with unassigned keys left","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java","lineNumber":175,"sourceCode":"        Maps.newHashMapWithExpectedSize(sortedStatistics.size());\n    Iterator<SortKey> mapKeyIterator = sortedStatistics.keySet().iterator();\n    int subtaskId = 0;\n    SortKey currentKey = null;\n    long keyRemainingWeight = 0L;\n    long subtaskRemainingWeight = targetWeightPerSubtask;\n    List<Integer> assignedSubtasks = Lists.newArrayList();\n    List<Long> subtaskWeights = Lists.newArrayList();\n    while (mapKeyIterator.hasNext() || currentKey != null) {\n      // This should never happen because target weight is calculated using ceil function.\n      if (subtaskId >= numPartitions) {\n        LOG.error(\n            \"Internal algorithm error: exhausted subtasks with unassigned keys left. number of partitions: {}, \"\n                + \"target weight per subtask: {}, close file cost in weight: {}, data statistics: {}\",\n            numPartitions,\n            targetWeightPerSubtask,\n            closeFileCostWeight,\n            sortedStatistics);\n        throw new IllegalStateException(\n            \"Internal algorithm error: exhausted subtasks with unassigned keys left\");\n      }\n\n      if (currentKey == null) {\n        currentKey = mapKeyIterator.next();\n        keyRemainingWeight = sortedStatistics.get(currentKey);\n      }\n\n      assignedSubtasks.add(subtaskId);\n      if (keyRemainingWeight < subtaskRemainingWeight) {\n        // assign the remaining weight of the key to the current subtask\n        subtaskWeights.add(keyRemainingWeight);\n        subtaskRemainingWeight -= keyRemainingWeight;\n        keyRemainingWeight = 0L;\n      } else {\n        // filled up the current subtask\n        long assignedWeight = subtaskRemainingWeight;\n        keyRemainingWeight -= subtaskRemainingWeight;","sourceCodeStart":157,"sourceCodeEnd":193,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java#L157-L193","documentation":"MapAssignment.buildAssignment() distributes partition keys across Flink writer subtasks using the collected key-weight statistics. The algorithm iterates sorted keys and subtasks; if it consumes every subtask while keys with remaining weight are still unassigned, the internal invariants are broken, so it throws IllegalStateException after logging the partitions, target weight, close-file cost weight, and statistics. This indicates a bug in the range-assignment algorithm rather than user input.","triggerScenarios":"Running the range-partitioned shuffle write path (sorted write with downstream shuffle) when the assignment loop exhausts subtask slots while mapKeyIterator still has keys — e.g. extreme distributions of key weights combined with very small/large close-file-cost weight settings.","commonSituations":"Highly skewed sort-key statistics (one key hugely heavier than targetWeightPerSubtask) or misconfigured write.parallelism vs. statistics granularity; also possible when numPartitions is degenerate (1 subtask) and key weight exceeds the target.","solutions":["Report the error with the logged statistics (partitions, targetWeightPerSubtask, closeFileCostWeight, sortedStatistics) to the Iceberg project — it is an internal invariant violation.","Increase write parallelism (write distribution mode hash/range parallelism) so the target weight per subtask is larger relative to single-key weights.","Adjust table property write.distribution.mode or disable the range shuffle (use hash/none) as a workaround.","Reduce skew by choosing a sort order with higher-cardinality leading columns."],"exampleFix":"// before\nALTER TABLE t SET ('write.distribution-mode'='range');\n// after (workaround while skew persists)\nALTER TABLE t SET ('write.distribution-mode'='hash');","handlingStrategy":"fallback","validationCode":"// check skew before enabling range shuffle: if maxKeyWeight > targetWeightPerSubtask, use hash mode","typeGuard":null,"tryCatchPattern":"try {\n  assignment = MapAssignment.buildAssignment(...);\n} catch (IllegalStateException e) {\n  LOG.warn(\"Range assignment failed; falling back to hash distribution\", e);\n  assignment = hashAssignment();\n}","preventionTips":["Choose sort orders with high-cardinality leading columns to limit key weight skew.","Keep write parallelism large enough that targetWeightPerSubtask exceeds any single key weight.","Report the logged diagnostics (partitions, target weight, statistics) to the Iceberg project."],"tags":["flink","shuffle","range-assignment","internal-error"],"backgroundTag":"internal-invariant-violation","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}