apache/beam · error · RuntimeException

UnboundedReader must override getCurrentRecordOffset() if of

Error message

UnboundedReader must override getCurrentRecordOffset() if offset-based deduplication is enabled for the UnboundedSource.

What it means

UnboundedReader.getCurrentRecordOffset() has no default implementation at all: it always throws RuntimeException. It must be overridden by any reader whose source participates in offset-based deduplication, since the runner uses the offset to detect replays. Unlike getCurrentRecordId (which is only checked when requiresDeduping() is true), any call into this default implementation is a bug in the reader implementation.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:230

     *
     * <p>This method has the same restrictions on when it can be called as {@link #getCurrent} and
     * {@link #getCurrentTimestamp}.
     *
     * @throws NoSuchElementException if the reader is at the beginning of the input and {@link
     *     #start} or {@link #advance} wasn't called, or if the last {@link #start} or {@link
     *     #advance} returned {@code false}.
     */
    public byte[] getCurrentRecordId() throws NoSuchElementException {
      if (getCurrentSource().requiresDeduping()) {
        throw new IllegalStateException(
            "getCurrentRecordId() must be overridden if requiresDeduping returns true()");
      }
      return EMPTY;
    }

    /* Returns the offset for the current record of this unbounded reader. */
    public byte[] getCurrentRecordOffset() {
      throw new RuntimeException(
          "UnboundedReader must override getCurrentRecordOffset() if offset-based deduplication is enabled for the UnboundedSource.");
    }

    /**
     * Returns a timestamp before or at the timestamps of all future elements read by this reader.
     *
     * <p>This can be approximate. If records are read that violate this guarantee, they will be
     * considered late, which will affect how they will be processed. See {@link
     * org.apache.beam.sdk.transforms.windowing.Window} for more information on late data and how to
     * handle it.
     *
     * <p>However, this value should be as late as possible. Downstream windows may not be able to
     * close until this watermark passes their end.
     *
     * <p>For example, a source may know that the records it reads will be in timestamp order. In
     * this case, the watermark can be the timestamp of the last record read. For a source that does
     * not have natural timestamps, timestamps can be set to the time of reading, in which case the
     * watermark is the current clock time.

View on GitHub (pinned to 12126d8942)

Solutions

  1. Override getCurrentRecordOffset() in your UnboundedReader and return the current record's offset as serialized bytes.
  2. Upgrade or patch the connector so its reader implements the offset API expected by the runner.
  3. If the runner should not query offsets for your source, verify source/reader configuration (e.g. requiresDeduping false) and runner version compatibility.

Example fix

// before
class MyReader extends UnboundedReader<MyRecord> {
  // inherits throwing getCurrentRecordOffset
}

// after
class MyReader extends UnboundedReader<MyRecord> {
  @Override
  public byte[] getCurrentRecordOffset() {
    return Longs.toByteArray(currentOffset);
  }
}
Defensive patterns

Strategy: validation

Validate before calling

checkState(!(reader instanceof UnboundedReader && overrideDeclaresGetCurrentRecordOffset(reader.getClass())), "reader must override getCurrentRecordOffset()");

Type guard

static boolean overridesGetCurrentRecordOffset(Class<?> c) {
  try { return !c.getMethod("getCurrentRecordOffset").getDeclaringClass().equals(UnboundedReader.class); }
  catch (NoSuchMethodException e) { return false; }
}

Try / catch

try {
  byte[] offset = reader.getCurrentRecordOffset();
} catch (RuntimeException e) {
  throw new UnsupportedOperationException("source does not support offset-based dedup", e);
}

Prevention

When it happens

Trigger: A runner or DoFn runner harness (e.g. via processElement/tryClaim paths such as Splittable DoFn wrapper or deduping runners) calls getCurrentRecordOffset() on an UnboundedReader subclass that did not override it — regardless of requiresDeduping(), any non-overriding reader that reaches this method throws.

Common situations: Implementing a custom unbounded source and missing the override; running a source on a Beam version where the runner newly queries record offsets; using a third-party connector written for an older Beam API without this method.

Related errors

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