{"record":{"id":"3841c3812effb140","repo":"apache/iceberg","slug":"failed-to-process-tasks-iterable","errorCode":null,"errorMessage":"Failed to process tasks iterable","messagePattern":"Failed to process tasks iterable","errorType":"exception","errorClass":"UncheckedIOException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSplitPlanner.java","lineNumber":68,"sourceCode":"      List<CombinedScanTask> tasks = Lists.newArrayList(tasksIterable);\n      FlinkInputSplit[] splits = new FlinkInputSplit[tasks.size()];\n      boolean exposeLocality = context.exposeLocality();\n\n      Tasks.range(tasks.size())\n          .stopOnFailure()\n          .executeWith(exposeLocality ? workerPool : null)\n          .run(\n              index -> {\n                CombinedScanTask task = tasks.get(index);\n                String[] hostnames = null;\n                if (exposeLocality) {\n                  hostnames = Util.blockLocations(table.io(), task);\n                }\n                splits[index] = new FlinkInputSplit(index, task, hostnames);\n              });\n      return splits;\n    } catch (IOException e) {\n      throw new UncheckedIOException(\"Failed to process tasks iterable\", e);\n    }\n  }\n\n  /** This returns splits for the FLIP-27 source */\n  public static List<IcebergSourceSplit> planIcebergSourceSplits(\n      Table table, ScanContext context, ExecutorService workerPool) {\n    try (CloseableIterable<CombinedScanTask> tasksIterable =\n        planTasks(table, context, workerPool)) {\n      return Lists.newArrayList(\n          CloseableIterable.transform(tasksIterable, IcebergSourceSplit::fromCombinedScanTask));\n    } catch (IOException e) {\n      throw new UncheckedIOException(\"Failed to process task iterable: \", e);\n    }\n  }\n\n  static CloseableIterable<CombinedScanTask> planTasks(\n      Table table, ScanContext context, ExecutorService workerPool) {\n    ScanMode scanMode = checkScanMode(context);","sourceCodeStart":50,"sourceCodeEnd":86,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSplitPlanner.java#L50-L86","documentation":"FlinkSplitPlanner.planInputSplits materializes CombinedScanTask tasks into FlinkInputSplits; IOException while consuming the tasks iterable (including block location lookups via table.io()) is wrapped in UncheckedIOException. It indicates IO failure while reading table metadata or computing hostnames during legacy FlinkSource split planning.","triggerScenarios":"Calling planInputSplits when table.io() cannot read manifests/metadata files, Util.blockLocations fails on the filesystem, or the CloseableIterable of tasks throws during iteration due to expired snapshots or unreachable storage.","commonSituations":"HDFS NameNode unavailability, S3 throttling/credentials errors during manifest reads, snapshot expiry race with compaction jobs, misconfigured IO implementation.","solutions":["Check the chained IOException cause for the underlying storage error","Verify storage credentials/connectivity for the configured FileIO","Ensure snapshot expiration does not race with split planning; pin snapshot-id if needed","Retry planning after transient storage errors; for S3 throttling, reduce workerPool parallelism"],"exampleFix":"// before\nFlinkInputSplit[] splits = FlinkSplitPlanner.planInputSplits(table, context, workerPool);\n// after: pin snapshot and check IO first\ntable.refresh();\nlong snapId = table.currentSnapshot().snapshotId();\nScanContext pinned = context.copyWithSnapshotId(snapId);\nFlinkInputSplit[] splits = FlinkSplitPlanner.planInputSplits(table, pinned, workerPool);","handlingStrategy":"retry","validationCode":"table.refresh();\nPreconditions.checkState(table.currentSnapshot() != null, \"No snapshot available for planning\");","typeGuard":null,"tryCatchPattern":"try {\n  splits = FlinkSplitPlanner.planInputSplits(table, context, pool);\n} catch (UncheckedIOException e) {\n  if (isTransient(e.getCause())) { splits = FlinkSplitPlanner.planInputSplits(table, context, pool); }\n  else throw e;\n}","preventionTips":["Check storage connectivity/credentials at job startup","Avoid racing snapshot expiration with split planning","Reduce workerPool size against throttling stores (S3)"],"tags":["flink","split-planning","io","filesystem"],"backgroundTag":"file-read-failed","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"}