{"record":{"id":"cf1adb64f4c1d9b6","repo":"apache/beam","slug":"spark-receiver-was-interrupted-before-polling-started","errorCode":null,"errorMessage":"Spark Receiver was interrupted before polling started","messagePattern":"Spark Receiver was interrupted before polling started","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":296,"sourceCode":"    Receiver<V> sparkReceiver;\n    try {\n      sparkReceiver = sparkReceiverBuilder.build();\n    } catch (Exception e) {\n      LOG.error(\"Can not build Spark Receiver\", e);\n      throw new IllegalStateException(\"Spark Receiver was not built!\");\n    }\n    LOG.debug(\"Restriction {}\", tracker.currentRestriction().toString());\n    sparkConsumer = new SparkConsumerWithOffset<>(tracker.currentRestriction().getFrom());\n    sparkConsumer.start(sparkReceiver);\n\n    Long recordsProcessed = 0L;\n    while (true) {\n      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","sourceCodeStart":278,"sourceCodeEnd":314,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java#L278-L314","documentation":"Before the first poll, processElement sleeps startPollTimeoutSec to give the Spark Receiver time to start producing records. If that sleep is interrupted, the DoFn converts the InterruptedException into this IllegalStateException, aborting element processing because the receiver never reached a pollable state.","triggerScenarios":"The worker thread waiting TimeUnit.SECONDS.sleep(startPollTimeoutSec) receives an interrupt — typically during pipeline drain, worker shutdown, or a runner cancelling a slow bundle.","commonSituations":"Autoscaler terminating workers mid-bundle; pipeline cancellation; excessive startPollTimeoutSec causing bundles to overlap shutdown windows; runner-enforced bundle timeouts interrupting the thread.","solutions":["Rerun the bundle/pipeline; the interruption is usually benign (shutdown) unless recurrent","Reduce startPollTimeoutSec so bundles finish before runner timeouts","Avoid cancelling/terminating workers while bundles are actively processing","If recurrent, check runner logs for what is interrupting threads (drain, autoscaling, timeouts)"],"exampleFix":"// before\n.withStartPollTimeout(120) // very long initial wait, vulnerable to interrupts\n// after\n.withStartPollTimeout(5)","handlingStrategy":"try-catch","validationCode":"if (options.getStartPollTimeoutSec() > 30) {\n  LOG.warn(\"Large startPollTimeout increases bundle interruption exposure\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  readBundle();\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"interrupted before polling\")) {\n    LOG.warn(\"Bundle interrupted during start-poll wait; safe to rerun if during shutdown\");\n  }\n}","preventionTips":["Keep startPollTimeoutSec small","Avoid terminating workers mid-bundle","Investigate recurring interrupts in runner logs","Schedule drains outside active processing windows"],"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"}