{"record":{"id":"6c63db43b4e496a3","repo":"apache/iceberg","slug":"queue-filling-failed-6c63db","errorCode":null,"errorMessage":"Queue filling failed","messagePattern":"Queue filling failed","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java","lineNumber":220,"sourceCode":"              }\n              // end offset may be synthetic and not exist in the queue\n              boolean endOffsetSynthetic =\n                  currentOffset.snapshotId() == endOffset.snapshotId()\n                      && (currentOffset.position() + 1) == endOffset.position();\n              shouldTerminate = endOffsetPeek || endOffsetSynthetic;\n            } else {\n              LOG.trace(\"planFiles hasn't reached {}, waiting\", endOffset);\n            }\n          } while (!shouldTerminate\n              && refreshFailedThrowable == null\n              && fillQueueFailedThrowable == null);\n\n          if (refreshFailedThrowable != null) {\n            throw new RuntimeException(\"Table refresh failed\", refreshFailedThrowable);\n          }\n\n          if (fillQueueFailedThrowable != null) {\n            throw new RuntimeException(\"Queue filling failed\", fillQueueFailedThrowable);\n          }\n\n          LOG.info(\n              \"completed planFiles for {}, startOffset: {}, endOffset: {}, files: {}, rows: {}\",\n              table().name(),\n              startOffset,\n              endOffset,\n              filesInPlan,\n              rowsInPlan);\n          return result;\n        });\n  }\n\n  /**\n   * This needs to be non destructive on the queue as spark could call this multiple times. Each\n   * time, depending on the table state it could return something different\n   *\n   * @param startOffset the starting offset of the next microbatch","sourceCodeStart":202,"sourceCodeEnd":238,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java#L202-L238","documentation":"In Spark structured streaming reads, AsyncSparkMicroBatchPlanner.planFiles plans batches on a background thread while the main thread runs concurrently. If queue-filling (loading manifests/files into the planning queue) threw, the throwable is captured and re-thrown on the main thread as 'Queue filling failed' with the original cause attached. The actual root cause is in the getCause() chain.","triggerScenarios":"Calling SparkStreamingCheckpoints/planFiles when the async planner's queue-filling step throws any exception (typically an IO error reading manifests via FileIO, a table refresh that swapped metadata concurrently, or an unavailable file system) while other errors are absent.","commonSituations":"Object store transient failures (S3/GCS throttling) during manifest listing; HDFS NameNode unavailability; table deleted or schema changed mid-stream; credentials expiring during long streams.","solutions":["Inspect the chained cause (e.getCause()) with full stack trace to find the real failure (usually IOException from manifest reads).","Verify the warehouse/FileIO credentials and network path to the table location are valid for the streaming job.","Retry/restart the query; transient object-store errors often resolve on restart.","Check that no concurrent table mutation (schema/version rewrite) corrupted the snapshot the planner is reading."],"exampleFix":"// before\n// generic wrapper hides root cause; enable cause inspection\ntry { batch = planFiles(...); } catch (RuntimeException e) {\n  log.error(\"planning failed\", e.getCause());\n}\n// after\n// inspect cause; add retry for transient FileIO errors\ntry { batch = planFiles(...); } catch (RuntimeException e) {\n  Throwable root = e.getCause();\n  if (isTransient(root)) retryWithBackoff(); else throw e;\n}","handlingStrategy":"retry","validationCode":"// pre-check table reachability before starting stream\nTable table = catalog.loadTable(ident);\ntable.refresh(); // throws early if FileIO/credentials are broken","typeGuard":null,"tryCatchPattern":"try { planBatch(); } catch (RuntimeException e) {\n  if (e.getCause() instanceof IOException && isTransient(e.getCause())) retryWithBackoff();\n  else throw e;\n}","preventionTips":["Validate FileIO credentials and warehouse connectivity before launching the stream.","Monitor object-store throttling metrics; pre-size request rates.","Avoid concurrent destructive table mutations during streaming reads."],"tags":["spark","streaming","async-planning","io"],"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-14T11:17:12.474Z"}