{"record":{"id":"b7362affcecf845d","repo":"apache/beam","slug":"checkpointmark-must-override-getoffsetlimit-if-offset-based","errorCode":null,"errorMessage":"CheckpointMark must override getOffsetLimit() if offset-based deduplication is enabled for the UnboundedSource.","messagePattern":"CheckpointMark must override getOffsetLimit\\(\\) if offset-based deduplication is enabled for the UnboundedSource\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java","lineNumber":161,"sourceCode":"     *       created still exists at the time this method is called.\n     * </ul>\n     */\n    void finalizeCheckpoint() throws IOException;\n\n    @SuppressWarnings(\"ClassInitializationDeadlock\")\n    NoopCheckpointMark NOOP_CHECKPOINT_MARK = new NoopCheckpointMark();\n\n    /** A checkpoint mark that does nothing when finalized. */\n    final class NoopCheckpointMark implements UnboundedSource.CheckpointMark {\n      @Override\n      public void finalizeCheckpoint() throws IOException {\n        // nothing to do\n      }\n    }\n\n    /* Get offset limit for unbounded source split checkpoint. */\n    default byte[] getOffsetLimit() {\n      throw new RuntimeException(\n          \"CheckpointMark must override getOffsetLimit() if offset-based deduplication is enabled for the UnboundedSource.\");\n    }\n  }\n\n  /**\n   * A {@code Reader} that reads an unbounded amount of input.\n   *\n   * <p>A given {@code UnboundedReader} object will only be accessed by a single thread at once.\n   */\n  public abstract static class UnboundedReader<OutputT> extends Source.Reader<OutputT> {\n    private static final byte[] EMPTY = new byte[0];\n\n    /**\n     * Initializes the reader and advances the reader to the first record. If the reader has been\n     * restored from a checkpoint then it should advance to the next unread record at the point the\n     * checkpoint was taken.\n     *\n     * <p>This method will be called exactly once. The invocation will occur prior to calling {@link","sourceCodeStart":143,"sourceCodeEnd":179,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java#L143-L179","documentation":"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().","triggerScenarios":"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.","commonSituations":"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.","solutions":["Override getOffsetLimit() in your CheckpointMark implementation and return the serialized offset limit bytes.","If your source does not need dedup, make requiresDeduping() return false so this method is never invoked.","Upgrade/patch the connector library whose CheckpointMark predates the dedup API."],"exampleFix":"// before\nclass MyCheckpoint implements UnboundedSource.CheckpointMark {\n  // no getOffsetLimit override\n}\n\n// after\nclass MyCheckpoint implements UnboundedSource.CheckpointMark {\n  @Override\n  public byte[] getOffsetLimit() {\n    return serializeOffset(limitOffset);\n  }\n}","handlingStrategy":"validation","validationCode":"if (source.requiresDeduping()) {\n  checkState(checkpointOverridesGetOffsetLimit(checkpointClass), \"CheckpointMark must override getOffsetLimit()\");\n}","typeGuard":"static boolean supportsOffsetLimit(Class<? extends UnboundedSource.CheckpointMark> c) {\n  try { c.getMethod(\"getOffsetLimit\"); return c != UnboundedSource.CheckpointMark.class; }\n  catch (NoSuchMethodException e) { return false; }\n}","tryCatchPattern":"try {\n  byte[] limit = checkpoint.getOffsetLimit();\n} catch (RuntimeException e) {\n  // disable dedup or upgrade connector implementing getOffsetLimit\n}","preventionTips":["Whenever requiresDeduping() returns true, implement getOffsetLimit in the CheckpointMark from day one.","Add a source-level unit test that calls getOffsetLimit and getCurrentRecordId/Offset on a real checkpoint.","Pin connector versions compatible with your Beam version's dedup API."],"tags":["java","apache-beam","io","unbounded-source","dedup"],"backgroundTag":"method-not-implemented","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}