{"record":{"id":"da09d76c821067e2","repo":"apache/iceberg","slug":"interrupted-while-polling-queue-da09d7","errorCode":null,"errorMessage":"Interrupted while polling queue","messagePattern":"Interrupted while polling queue","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"warning","filePath":"spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java","lineNumber":183,"sourceCode":"        key -> {\n          LOG.info(\n              \"running planFiles for {}, startOffset: {}, endOffset: {}\",\n              table().name(),\n              startOffset,\n              endOffset);\n          List<FileScanTask> result = new LinkedList<>();\n          Pair<StreamingOffset, FileScanTask> elem;\n          StreamingOffset currentOffset;\n          boolean shouldTerminate = false;\n          long filesInPlan = 0;\n          long rowsInPlan = 0;\n\n          do {\n            try {\n              elem = queue.pollFirst(QUEUE_POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS);\n            } catch (InterruptedException e) {\n              Thread.currentThread().interrupt();\n              throw new RuntimeException(\"Interrupted while polling queue\", e);\n            }\n\n            if (elem != null) {\n              currentOffset = elem.first();\n              LOG.debug(\"planFiles consumed: {}\", currentOffset);\n              FileScanTask currentTask = elem.second();\n              filesInPlan += 1;\n              long elemRows = currentTask.file().recordCount();\n              rowsInPlan += elemRows;\n              queuedFileCount.decrementAndGet();\n              queuedRowCount.addAndGet(-elemRows);\n              result.add(currentTask);\n\n              // try to peek at the next entry of the queue and see if we should stop\n              Pair<StreamingOffset, FileScanTask> nextElem = queue.peekFirst();\n              boolean endOffsetPeek = false;\n              if (nextElem != null) {\n                endOffsetPeek = endOffset.equals(nextElem.first());","sourceCodeStart":165,"sourceCodeEnd":201,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/AsyncSparkMicroBatchPlanner.java#L165-L201","documentation":"AsyncSparkMicroBatchPlanner.planFiles consumes planned (offset, FileScanTask) pairs from a queue filled by a background thread, polling with a timeout in a loop. If the polling thread is interrupted (Thread.interrupt from task cancellation or query stop), it restores the interrupt flag and wraps the InterruptedException in a RuntimeException with this message.","triggerScenarios":"A Spark structured streaming query reading an Iceberg table via the async micro-batch planner is cancelled or stopped while planFiles is blocked waiting on the queue; Spark's interrupt handling calls Thread.interrupt during query termination or executor shutdown.","commonSituations":"streamingQuery.stop() racing with an in-flight micro-batch; job cancellation from a scheduler timeout; executor shutdown/preemption in YARN/K8s while planning; Ctrl-C on a streaming query in a notebook.","solutions":["Usually benign during query termination — verify the streaming query was intentionally stopped or cancelled.","If unexpected, check for code calling Thread.interrupt() on Spark task threads (custom listeners, watchdogs).","Retry/resume the streaming query from the last committed checkpoint; offsets are durable.","If it recurs without cancellation, report/inspect the background planner thread for deadlock keeping the queue empty."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    query.processAllAvailable();\n} catch (RuntimeException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"Interrupted while polling queue\")) {\n        // expected during stop(); re-interrupt and exit cleanly\n        Thread.currentThread().interrupt();\n    }\n}","preventionTips":["Always stop streaming queries via streamingQuery.stop() and treat interrupt-based exceptions as normal shutdown","Avoid custom code that interrupts Spark execution threads","Rely on checkpoints so interrupted queries resume cleanly"],"tags":["spark","streaming","interruption"],"backgroundTag":"request-timeout","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"}