apache/beam · error · IllegalStateException

Spark Receiver was interrupted while waiting to poll new…

Error message

Spark Receiver was interrupted while waiting to poll new records

What it means

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.

Solutions

1. Rerun the pipeline; single interruptions during shutdown are expected
2. Lower pullFrequencySec to reduce time spent in interruptible sleeps
3. Coordinate pipeline shutdowns so workers are not killed mid-bundle
4. Investigate the interrupting component in runner logs if it occurs without a shutdown

Example fix

// before
.withPullFrequency(300) // long pacing sleep
// after
.withPullFrequency(10)

if (options.getPullFrequencySec() > 60) {
  LOG.warn("Large pullFrequencySec extends interruptible sleep windows");
}

try {
  readBundle();
} catch (IllegalStateException e) {
  if (e.getMessage().contains("interrupted while waiting to poll")) {
    LOG.warn("Bundle interrupted during pull-frequency wait; rerun if caused by shutdown");
  }
}

When it happens

Trigger: TimeUnit.SECONDS.sleep(pullFrequencySec) interrupted — pipeline cancellation/drain, worker shutdown, or runner-initiated thread interruption while the DoFn waits to poll for new records.

Common situations: Pipeline drains or cluster autoscaling during long-running streaming reads; very large pullFrequencySec values extending exposure to shutdowns; deliberate cancellation by an operator.

Related errors

• Spark Receiver was interrupted before polling started
• Spark Receiver was not built!
• Spark Receiver was not initialized
• A list of URNs for overriding transforms was provided but…
• A cannot be expanded

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/829b96ddb5e3b217. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java:309

      LOG.debug("Start polling records");
      try {
        TimeUnit.SECONDS.sleep(startPollTimeoutSec);
      } catch (InterruptedException e) {
        LOG.error("SparkReceiver was interrupted before polling started", e);
        throw new IllegalStateException("Spark Receiver was interrupted before polling started");
      }
      if (!sparkConsumer.hasRecords()) {
        LOG.debug("No records left");
        ((HasOffset) sparkReceiver).setCheckpoint(recordsProcessed);
        sparkConsumer.stop();
        tracker.checkDone();
        if (pullFrequencySec != 0L) {
          LOG.debug("Waiting to poll for new records...");
          try {
            TimeUnit.SECONDS.sleep(pullFrequencySec);
          } catch (InterruptedException e) {
            LOG.error("SparkReceiver was interrupted while waiting to poll new records", e);
            throw new IllegalStateException(
                "Spark Receiver was interrupted while waiting to poll new records");
          }
        }
        OffsetRange currentRestriction = tracker.currentRestriction();
        if (currentRestriction != null
            && currentRestriction.getFrom() == currentRestriction.getTo()) {
          LOG.info("Stop for empty restriction: {}", currentRestriction);
          return ProcessContinuation.stop();
        } else {
          LOG.info("Resume for restriction: {}", currentRestriction);
          return ProcessContinuation.resume();
        }
      }
      while (sparkConsumer.hasRecords()) {
        V record = sparkConsumer.poll();
        if (record != null) {
          Long offset = getOffsetFn.apply(record);
          if (!tracker.tryClaim(offset)) {

View on GitHub (pinned to 12126d8942)