apache/beam · error · UnsupportedOperationException

Given Spark Receiver class

Error message

Given Spark Receiver class %s doesn't implement HasOffset interface, therefore it is not supported!

What it means

SparkReceiverIO's Read with offsets mode only supports Spark Receiver classes that implement the HasOffset interface, because offset tracking for restartability requires the receiver to expose its current offset. During pipeline expansion (expand), the transform checks the receiver class supplied via withSparkReceiverBuilder() and refuses to run an incompatible receiver rather than fail later at runtime.

Solutions

Make the receiver class implement the HasOffset interface (add getCurrentOffset returning the receiver's current offset and update it as records arrive).
If offset tracking is genuinely not possible, use SparkReceiverIO.read() (streaming without offsets) instead of readWithOffsets().
Verify the class passed to ReceiverBuilder is the concrete receiver that implements HasOffset, not a wrapper or a wrong generic parameter.

Example fix

// before
return io.apply(SparkReceiverIO.<Long>readWithOffsets()
    .withSparkReceiverBuilder(new ReceiverBuilder<>(MyReceiver.class)));
// after
public class MyReceiver extends Receiver<Long> implements HasOffset {
  private long currentOffset = 0;
  @Override
  public long getCurrentOffset() { return currentOffset; }
  // ... update currentOffset in onStart/onStop/receive
}

Defensive patterns

Strategy: validation

Validate before calling

// Java — before building the read
Class<? extends Receiver<Long>> rc = myReceiverClass;
if (!HasOffset.class.isAssignableFrom(rc)) {
  throw new IllegalStateException(rc.getName() + " must implement HasOffset for readWithOffsets()");
}

Type guard

boolean supportsOffsets(Class<? extends Receiver<?>> c) { return HasOffset.class.isAssignableFrom(c); }

Prevention

Implement HasOffset on every custom receiver used with readWithOffsets().
Keep receiver classes for offset mode and plain mode clearly named/separated.
Add a unit test that asserts HasOffset.class.isAssignableFrom(yourReceiver.class).

When it happens

Trigger: Calling SparkReceiverIO.<T>readWithOffsets().withSparkReceiverBuilder(new ReceiverBuilder<>(SomeReceiver.class)) where SomeReceiver (or a subclass of it) does not implement org.apache.beam.sdk.io.sparkreceiver.HasOffset; the exception is thrown when the pipeline graph is constructed/expanded.

Common situations: Using a custom Spark Receiver written for the plain SparkReceiverIO.read() (CustomReceiverWithOffset-free) path and then switching to readWithOffsets(); copying an example receiver from older Beam versions before HasOffset was introduced; forgetting to add offset reporting methods when upgrading the pipeline.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

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

Appendix: source

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

      checkStateNotNull(getGetOffsetFn(), "withGetOffsetFn() is required");
    }
  }

  static class ReadFromSparkReceiverViaSdf<V> extends PTransform<PBegin, PCollection<V>> {

    private final Read<V> sparkReceiverRead;

    ReadFromSparkReceiverViaSdf(Read<V> sparkReceiverRead) {
      this.sparkReceiverRead = sparkReceiverRead;
    }

    @Override
    public PCollection<V> expand(PBegin input) {
      final ReceiverBuilder<V, ? extends Receiver<V>> sparkReceiverBuilder =
          sparkReceiverRead.getSparkReceiverBuilder();
      checkStateNotNull(sparkReceiverBuilder, "withSparkReceiverBuilder() is required");
      if (!HasOffset.class.isAssignableFrom(sparkReceiverBuilder.getSparkReceiverClass())) {
        throw new UnsupportedOperationException(
            String.format(
                "Given Spark Receiver class %s doesn't implement HasOffset interface,"
                    + " therefore it is not supported!",
                sparkReceiverBuilder.getSparkReceiverClass().getName()));
      } else {
        LOG.info("{} started reading", ReadFromSparkReceiverWithOffsetDoFn.class.getSimpleName());
        return input
            .apply(Impulse.create())
            .apply(ParDo.of(new ReadFromSparkReceiverWithOffsetDoFn<>(sparkReceiverRead)));
        // TODO: Split data from SparkReceiver into multiple workers
      }
    }
  }
}

View on GitHub (pinned to 12126d8942)