{"record":{"id":"eda9aaf3c0495e58","repo":"apache/iceberg","slug":"internal-algorithm-error-exhausted-subtasks-with","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/v1.20/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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java#L157-L193","documentation":"MapAssignment.buildAssignment distributes map-key weights across subtasks proportionally to data statistics. If the algorithm consumes all subtasks while keys with unassigned weight remain, it logs the algorithm parameters and throws IllegalStateException('Internal algorithm error: exhausted subtasks with unassigned keys left'). This is an invariant failure of the assignment algorithm, not a user configuration error per se.","triggerScenarios":"During assignment computation, the loop runs out of subtask slots while sortedStatistics still holds keys with remaining weight — degenerate inputs such as zero/negative target weight per subtask, extreme weight distributions, or bad data statistics can drive the algorithm into this state.","commonSituations":"Very skewed key distributions combined with small write parallelism; closeFileCost weighting configurations producing a target weight per subtask that cannot cover remaining keys; write parallelism changed drastically (rescale) with stale statistics.","solutions":["Increase the write (range-partition) parallelism so target weight per subtask is not degenerately small relative to total key weight","Inspect the logged values (numPartitions, targetWeightPerSubtask, closeFileCostInWeight) for zero/invalid values and fix the causing configuration","Refresh data statistics by running a few checkpoints so the assignment is recomputed from current distributions","Collect the full log context and file an Iceberg issue — this indicates an algorithm bug if inputs look sane"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"long totalWeight = sortedStatistics.values().stream().mapToLong(Long::longValue).sum();\nPreconditions.checkState(numPartitions > 0 && totalWeight > 0, \"Invalid inputs for map assignment: partitions=%s weight=%s\", numPartitions, totalWeight);","typeGuard":null,"tryCatchPattern":"try {\n  MapAssignment a = MapAssignment.assignment(subtasks, closeFileCostWeight, stats);\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"exhausted subtasks\")) {\n    // fall back to even distribution or recompute with fresh statistics\n  }\n  throw e;\n}","preventionTips":["Keep write parallelism proportional to data volume","Avoid extreme closeFileCost weight settings","Let a checkpoint refresh statistics before relying on assignments after heavy rescaling","Report reproducible failures with the logged algorithm parameters to Iceberg"],"tags":["flink","range-partitioning","algorithm","skew"],"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"}