apache/beam · error · IllegalStateException

Spark Receiver was interrupted before polling started

Error message

Spark Receiver was interrupted before polling started

What it means

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.

Solutions

1. Rerun the bundle/pipeline; the interruption is usually benign (shutdown) unless recurrent
2. Reduce startPollTimeoutSec so bundles finish before runner timeouts
3. Avoid cancelling/terminating workers while bundles are actively processing
4. If recurrent, check runner logs for what is interrupting threads (drain, autoscaling, timeouts)

Example fix

// before
.withStartPollTimeout(120) // very long initial wait, vulnerable to interrupts
// after
.withStartPollTimeout(5)

if (options.getStartPollTimeoutSec() > 30) {
  LOG.warn("Large startPollTimeout increases bundle interruption exposure");
}

try {
  readBundle();
} catch (IllegalStateException e) {
  if (e.getMessage().contains("interrupted before polling")) {
    LOG.warn("Bundle interrupted during start-poll wait; safe to rerun if during shutdown");
  }
}

When it happens

Trigger: The worker thread waiting TimeUnit.SECONDS.sleep(startPollTimeoutSec) receives an interrupt — typically during pipeline drain, worker shutdown, or a runner cancelling a slow bundle.

Common situations: Autoscaler terminating workers mid-bundle; pipeline cancellation; excessive startPollTimeoutSec causing bundles to overlap shutdown windows; runner-enforced bundle timeouts interrupting the thread.

Related errors

• Spark Receiver was interrupted while waiting to poll new…
• 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/cf1adb64f4c1d9b6. Report an issue: GitHub.

Appendix: source

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

    Receiver<V> sparkReceiver;
    try {
      sparkReceiver = sparkReceiverBuilder.build();
    } catch (Exception e) {
      LOG.error("Can not build Spark Receiver", e);
      throw new IllegalStateException("Spark Receiver was not built!");
    }
    LOG.debug("Restriction {}", tracker.currentRestriction().toString());
    sparkConsumer = new SparkConsumerWithOffset<>(tracker.currentRestriction().getFrom());
    sparkConsumer.start(sparkReceiver);

    Long recordsProcessed = 0L;
    while (true) {
      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

View on GitHub (pinned to 12126d8942)