{"record":{"id":"0a5473d91db1372a","repo":"apache/iceberg","slug":"internal-algorithm-error-exhausted-subtasks-with-0a5473","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.3/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.3/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java#L157-L193","documentation":"MapAssignment.buildAssignment() throws IllegalStateException when the range-assignment algorithm runs out of subtasks while map keys (partitions) still have unassigned weight. The algorithm should always have enough subtask capacity to place every key's weight; hitting this means the loop terminated early, which is treated as an internal invariant violation.","triggerScenarios":"Invoked via assignment() during MapRangePartitioner planning when sorted key statistics are inconsistent with the computed targetWeightPerSubtask / closeFileCostWeight — e.g., extreme key-weight skew, integer overflow in weights, or numPartitions reduced to an unexpectedly small value.","commonSituations":"Very skewed data statistics where one or few partitions dominate weight; write parallelism drastically reduced between runs; a genuine bug in the weighting math surfaced by an unusual statistics distribution.","solutions":["Check the write parallelism (sink subtask count) — a pathologically small value relative to the number of partitions can starve the algorithm; increase it.","Inspect the logged diagnostics (numPartitions, targetWeightPerSubtask, closeFileCostWeight, sortedStatistics) for overflow or skew.","Increase write.task.max / sink parallelism so target weight per subtask is not degenerate.","If reproducible with normal inputs, report it as a bug with the logged statistics — it signals an algorithm defect, not user error."],"exampleFix":"// before\nenv.from(...).sinkTo(icebergSink); // parallelism 1\n// after\nsinkBuilder = IcebergSink.builder()...;\nenv.from(...).sinkTo(icebergSink).setParallelism(numPartitions);","handlingStrategy":"validation","validationCode":"if (sinkParallelism < 1 || sinkParallelism > numPartitions * 4) {\n  throw new IllegalArgumentException(\"Suspicious sink parallelism: \" + sinkParallelism);\n}","typeGuard":null,"tryCatchPattern":"try {\n  assignment = MapAssignment.buildAssignment(...);\n} catch (IllegalStateException e) {\n  LOG.error(\"Assignment algorithm failed; check logged diagnostics\", e);\n  throw e;\n}","preventionTips":["Set sink parallelism proportional to partition count","Avoid extreme weight skew inputs; review logged sortedStatistics","Report reproducible failures as upstream bugs with diagnostics"],"tags":["flink","algorithm","invariant","range-partitioning"],"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-14T16:17:12.679Z"}