{"record":{"id":"25580c961cace0c1","repo":"apache/iceberg","slug":"internal-algorithm-error-exhausted-subtasks-with-25580c","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.1/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.1/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapAssignment.java#L157-L193","documentation":"MapAssignment.buildAssignment distributes map keys across subtasks by weight; the algorithm should always be able to place remaining key weight on unassigned subtasks. Reaching this IllegalStateException means the iteration exhausted subtasks while keys with unassigned weight remained — an internal invariant violation in the assignment algorithm.","triggerScenarios":"Hitting the weight-distribution loop with degenerate statistics inputs (e.g. extreme weight ratios, zero/negative target weight per subtask derived from closeFileCostWeight or huge key counts).","commonSituations":"Pathological collected data statistics during adaptive range shuffle (extremely skewed keys or tiny target weights); normally indicates a connector algorithm bug.","solutions":["File a bug with the logged diagnostics (numPartitions, targetWeightPerSubtask, closeFileCostWeight, statistics).","Disable the adaptive/shuffle-based write distribution (use write.distribution-mode=none or hash) as a workaround.","Upgrade the connector to a version with assignment algorithm fixes."],"exampleFix":"// workaround: avoid the adaptive range shuffle\nALTER TABLE t UNSET TBLPROPERTIES ('write.distribution-mode');","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  MapAssignment a = MapAssignment.assignment(numPartitions, closeFileCostWeight, statistics);\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"exhausted subtasks\")) {\n    LOG.error(\"Assignment algorithm bug; falling back to round-robin partitioning\", e);\n  } else throw e;\n}","preventionTips":["Disable shuffle-statistics-based distribution if you hit this (use write.distribution-mode=none or hash).","Report the logged diagnostics (partitions, target weight, statistics) to the connector project.","Upgrade to the latest connector release."],"tags":["flink","shuffle","algorithm"],"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"}