apache/beam · error · RuntimeException

CheckpointMark must override getOffsetLimit() if offset-base

Error message

CheckpointMark must override getOffsetLimit() if offset-based deduplication is enabled for the UnboundedSource.

What it means

UnboundedSource.CheckpointMark.getOffsetLimit() is a default method that exists purely to support offset-based deduplication of records from unbounded sources. If a runner enables deduping and queries the checkpoint's offset limit, the default implementation throws RuntimeException because only the source's own CheckpointMark subclass knows its offset representation. Custom CheckpointMark implementations must override it when their source declares requiresDeduping().

Source

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

     *       created still exists at the time this method is called.
     * </ul>
     */
    void finalizeCheckpoint() throws IOException;

    @SuppressWarnings("ClassInitializationDeadlock")
    NoopCheckpointMark NOOP_CHECKPOINT_MARK = new NoopCheckpointMark();

    /** A checkpoint mark that does nothing when finalized. */
    final class NoopCheckpointMark implements UnboundedSource.CheckpointMark {
      @Override
      public void finalizeCheckpoint() throws IOException {
        // nothing to do
      }
    }

    /* Get offset limit for unbounded source split checkpoint. */
    default byte[] getOffsetLimit() {
      throw new RuntimeException(
          "CheckpointMark must override getOffsetLimit() if offset-based deduplication is enabled for the UnboundedSource.");
    }
  }

  /**
   * A {@code Reader} that reads an unbounded amount of input.
   *
   * <p>A given {@code UnboundedReader} object will only be accessed by a single thread at once.
   */
  public abstract static class UnboundedReader<OutputT> extends Source.Reader<OutputT> {
    private static final byte[] EMPTY = new byte[0];

    /**
     * Initializes the reader and advances the reader to the first record. If the reader has been
     * restored from a checkpoint then it should advance to the next unread record at the point the
     * checkpoint was taken.
     *
     * <p>This method will be called exactly once. The invocation will occur prior to calling {@link

View on GitHub (pinned to 12126d8942)

Solutions

1. Override getOffsetLimit() in your CheckpointMark implementation and return the serialized offset limit bytes.
2. If your source does not need dedup, make requiresDeduping() return false so this method is never invoked.
3. Upgrade/patch the connector library whose CheckpointMark predates the dedup API.

Example fix

// before
class MyCheckpoint implements UnboundedSource.CheckpointMark {
  // no getOffsetLimit override
}

// after
class MyCheckpoint implements UnboundedSource.CheckpointMark {
  @Override
  public byte[] getOffsetLimit() {
    return serializeOffset(limitOffset);
  }
}

if (source.requiresDeduping()) {
  checkState(checkpointOverridesGetOffsetLimit(checkpointClass), "CheckpointMark must override getOffsetLimit()");
}

static boolean supportsOffsetLimit(Class<? extends UnboundedSource.CheckpointMark> c) {
  try { c.getMethod("getOffsetLimit"); return c != UnboundedSource.CheckpointMark.class; }
  catch (NoSuchMethodException e) { return false; }
}

try {
  byte[] limit = checkpoint.getOffsetLimit();
} catch (RuntimeException e) {
  // disable dedup or upgrade connector implementing getOffsetLimit
}

When it happens

Trigger: An UnboundedSource whose requiresDeduping() returns true uses a CheckpointMark that does not override getOffsetLimit(); a runner (e.g. Dataflow/Flink with dedup enabled) calls offsetLimit(...) on that checkpoint during finalize/commit and hits the default throwing implementation.

Common situations: Migrating a custom unbounded source to a runner that enables offset-based dedup; upgrading Beam where dedup APIs (getOffsetLimit/getCurrentRecordOffset) became required; third-party connectors (e.g. old Kafka/PubSub connectors) with CheckpointMarks written before these methods existed.

Related errors

• getCurrentRecordId() must be overridden if requiresDeduping
• UnboundedReader must override getCurrentRecordOffset() if of
• Failed closing channel to %s
• Error determining if %s allows dynamic splitting
• File spec %s not found

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