apache/iceberg · error · IllegalStateException

Internal algorithm error: exhausted subtasks with unassigned

Error message

Internal algorithm error: exhausted subtasks with unassigned keys left

What it means

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.

Source

Thrown at flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java:175

        Maps.newHashMapWithExpectedSize(sortedStatistics.size());
    Iterator<SortKey> mapKeyIterator = sortedStatistics.keySet().iterator();
    int subtaskId = 0;
    SortKey currentKey = null;
    long keyRemainingWeight = 0L;
    long subtaskRemainingWeight = targetWeightPerSubtask;
    List<Integer> assignedSubtasks = Lists.newArrayList();
    List<Long> subtaskWeights = Lists.newArrayList();
    while (mapKeyIterator.hasNext() || currentKey != null) {
      // This should never happen because target weight is calculated using ceil function.
      if (subtaskId >= numPartitions) {
        LOG.error(
            "Internal algorithm error: exhausted subtasks with unassigned keys left. number of partitions: {}, "
                + "target weight per subtask: {}, close file cost in weight: {}, data statistics: {}",
            numPartitions,
            targetWeightPerSubtask,
            closeFileCostWeight,
            sortedStatistics);
        throw new IllegalStateException(
            "Internal algorithm error: exhausted subtasks with unassigned keys left");
      }

      if (currentKey == null) {
        currentKey = mapKeyIterator.next();
        keyRemainingWeight = sortedStatistics.get(currentKey);
      }

      assignedSubtasks.add(subtaskId);
      if (keyRemainingWeight < subtaskRemainingWeight) {
        // assign the remaining weight of the key to the current subtask
        subtaskWeights.add(keyRemainingWeight);
        subtaskRemainingWeight -= keyRemainingWeight;
        keyRemainingWeight = 0L;
      } else {
        // filled up the current subtask
        long assignedWeight = subtaskRemainingWeight;
        keyRemainingWeight -= subtaskRemainingWeight;

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Report the error with the logged statistics (partitions, targetWeightPerSubtask, closeFileCostWeight, sortedStatistics) to the Iceberg project — it is an internal invariant violation.
2. Increase write parallelism (write distribution mode hash/range parallelism) so the target weight per subtask is larger relative to single-key weights.
3. Adjust table property write.distribution.mode or disable the range shuffle (use hash/none) as a workaround.
4. Reduce skew by choosing a sort order with higher-cardinality leading columns.

Example fix

// before
ALTER TABLE t SET ('write.distribution-mode'='range');
// after (workaround while skew persists)
ALTER TABLE t SET ('write.distribution-mode'='hash');

// check skew before enabling range shuffle: if maxKeyWeight > targetWeightPerSubtask, use hash mode

try {
  assignment = MapAssignment.buildAssignment(...);
} catch (IllegalStateException e) {
  LOG.warn("Range assignment failed; falling back to hash distribution", e);
  assignment = hashAssignment();
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• Unexpected command type in keyed stream: <cmd.type()>
• Unexpected ContentScanTask type: <task.getClass().getName()>
• Invalid operator event type: ${eventType}
• Internal algorithm error: exhausted subtasks with unassigned
• Invalid statistics type: ${type}. Should be Map or Sketch

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