apache/beam · error · IllegalStateException

Spark Receiver was not built!

Error message

Spark Receiver was not built!

What it means

In processElement, the DoFn builds a fresh Spark Receiver instance per element via the user-supplied SparkReceiverBuilder. If build() throws, the receiver cannot be created and this IllegalStateException aborts processing of the element.

Solutions

1. Inspect the logged "Can not build Spark Receiver" stack trace for the root cause
2. Make the builder lambda static/stateless and serializable, sourcing any config from pipeline options
3. Ensure the Receiver's constructor has no worker-environment-dependent failures (ports, files, credentials)
4. Test the receiver's no-arg construction outside Beam to confirm it builds cleanly

Example fix

// before
.withSparkReceiverBuilder(() -> new MyReceiver(nonSerializableHelper.getConfig()))
// after
.withSparkReceiverBuilder(() -> new MyReceiver(options.getReceiverParam()))

// preflight outside pipeline
Receiver<V> r = sparkReceiverBuilder.build(); // must not throw

boolean buildsCleanly(Supplier<Receiver<V>> b) { try { return b.get() != null; } catch (Exception e) { return false; } }

try {
  processElement();
} catch (IllegalStateException e) {
  if (e.getMessage().contains("Spark Receiver was not built")) {
    LOG.error("Receiver builder failed on worker; check serialization and env", e);
  }
  throw e;
}

When it happens

Trigger: sparkReceiverBuilder.build() throws for the given element — user-supplied lambda fails (bad constructor args, missing external resources, unsupported receiver state), or the receiver requires context not present on the Beam worker.

Common situations: Builder lambdas capturing non-serializable or worker-unavailable state; receivers that open network connections in their constructor and fail due to environment/firewall; bugs in custom Receiver constructors.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

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

  @ProcessElement
  public ProcessContinuation processElement(
      @Element byte[] element,
      RestrictionTracker<OffsetRange, Long> tracker,
      WatermarkEstimator<Instant> watermarkEstimator,
      OutputReceiver<V> receiver) {

    if (tracker.currentRestriction() != null) {
      LOG.info(
          "Start processing element. Restriction = {}", tracker.currentRestriction().toString());
    }
    SparkConsumer<V> sparkConsumer;
    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();

View on GitHub (pinned to 12126d8942)