{"record":{"id":"829b96ddb5e3b217","repo":"apache/beam","slug":"spark-receiver-was-interrupted-while-waiting-to-poll-new","errorCode":null,"errorMessage":"Spark Receiver was interrupted while waiting to poll new records","messagePattern":"Spark Receiver was interrupted while waiting to poll new records","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java","lineNumber":309,"sourceCode":"      LOG.debug(\"Start polling records\");\n      try {\n        TimeUnit.SECONDS.sleep(startPollTimeoutSec);\n      } catch (InterruptedException e) {\n        LOG.error(\"SparkReceiver was interrupted before polling started\", e);\n        throw new IllegalStateException(\"Spark Receiver was interrupted before polling started\");\n      }\n      if (!sparkConsumer.hasRecords()) {\n        LOG.debug(\"No records left\");\n        ((HasOffset) sparkReceiver).setCheckpoint(recordsProcessed);\n        sparkConsumer.stop();\n        tracker.checkDone();\n        if (pullFrequencySec != 0L) {\n          LOG.debug(\"Waiting to poll for new records...\");\n          try {\n            TimeUnit.SECONDS.sleep(pullFrequencySec);\n          } catch (InterruptedException e) {\n            LOG.error(\"SparkReceiver was interrupted while waiting to poll new records\", e);\n            throw new IllegalStateException(\n                \"Spark Receiver was interrupted while waiting to poll new records\");\n          }\n        }\n        OffsetRange currentRestriction = tracker.currentRestriction();\n        if (currentRestriction != null\n            && currentRestriction.getFrom() == currentRestriction.getTo()) {\n          LOG.info(\"Stop for empty restriction: {}\", currentRestriction);\n          return ProcessContinuation.stop();\n        } else {\n          LOG.info(\"Resume for restriction: {}\", currentRestriction);\n          return ProcessContinuation.resume();\n        }\n      }\n      while (sparkConsumer.hasRecords()) {\n        V record = sparkConsumer.poll();\n        if (record != null) {\n          Long offset = getOffsetFn.apply(record);\n          if (!tracker.tryClaim(offset)) {","sourceCodeStart":291,"sourceCodeEnd":327,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java#L291-L327","documentation":"Between polling rounds, processElement sleeps pullFrequencySec to pace record consumption. An interrupt during this wait is converted into this IllegalStateException, stopping the bundle since the pacing loop cannot continue safely.","triggerScenarios":"TimeUnit.SECONDS.sleep(pullFrequencySec) interrupted — pipeline cancellation/drain, worker shutdown, or runner-initiated thread interruption while the DoFn waits to poll for new records.","commonSituations":"Pipeline drains or cluster autoscaling during long-running streaming reads; very large pullFrequencySec values extending exposure to shutdowns; deliberate cancellation by an operator.","solutions":["Rerun the pipeline; single interruptions during shutdown are expected","Lower pullFrequencySec to reduce time spent in interruptible sleeps","Coordinate pipeline shutdowns so workers are not killed mid-bundle","Investigate the interrupting component in runner logs if it occurs without a shutdown"],"exampleFix":"// before\n.withPullFrequency(300) // long pacing sleep\n// after\n.withPullFrequency(10)","handlingStrategy":"try-catch","validationCode":"if (options.getPullFrequencySec() > 60) {\n  LOG.warn(\"Large pullFrequencySec extends interruptible sleep windows\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  readBundle();\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"interrupted while waiting to poll\")) {\n    LOG.warn(\"Bundle interrupted during pull-frequency wait; rerun if caused by shutdown\");\n  }\n}","preventionTips":["Use modest pullFrequencySec values","Coordinate shutdowns with pipeline activity","Check autoscaling policies to avoid mid-bundle kills","Restore/interrupt handling: rerun bundles after cancellation"],"tags":["java","beam","spark","receiver","interrupted"],"backgroundTag":"thread-interrupted","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}