{"record":{"id":"597b253cb995a74b","repo":"apache/beam","slug":"kafkacheckpointmark-reader-is-not-present-while-calling","errorCode":null,"errorMessage":"KafkaCheckpointMark reader is not present while calling getOffsetLimit().","messagePattern":"KafkaCheckpointMark reader is not present while calling getOffsetLimit\\(\\)\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java","lineNumber":75,"sourceCode":"\n  @Override\n  public void finalizeCheckpoint() {\n    reader.ifPresent(r -> r.finalizeCheckpointMarkAsync(this));\n    // Is it ok to commit asynchronously, or should we wait till this (or newer) is committed?\n    // Often multiple marks would be finalized at once, since we only need to finalize the latest,\n    // it is better to wait a little while. Currently maximum delay is same as KAFKA_POLL_TIMEOUT\n    // in the reader (1 second).\n  }\n\n  @Override\n  public String toString() {\n    return \"KafkaCheckpointMark{partitions=\" + Joiner.on(\",\").join(partitions) + '}';\n  }\n\n  @Override\n  public byte[] getOffsetLimit() {\n    if (!reader.isPresent()) {\n      throw new RuntimeException(\n          \"KafkaCheckpointMark reader is not present while calling getOffsetLimit().\");\n    }\n    if (!reader.get().offsetBasedDeduplicationSupported()) {\n      throw new RuntimeException(\n          \"Unexpected getOffsetLimit() called while KafkaUnboundedReader not configured for offset deduplication.\");\n    }\n\n    // KafkaUnboundedSource.split() must produce a 1:1 partition to split ratio.\n    checkState(partitions.size() == OFFSET_DEDUP_PARTITIONS_PER_SPLIT);\n    PartitionMark partition = partitions.get(/* index= */ 0);\n    return KafkaIOUtils.OffsetBasedDeduplication.encodeOffset(partition.getNextOffset());\n  }\n\n  /**\n   * A tuple to hold topic, partition, and offset that comprise the checkpoint for a single\n   * partition.\n   */\n  public static class PartitionMark implements Serializable {","sourceCodeStart":57,"sourceCodeEnd":93,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java#L57-L93","documentation":"KafkaCheckpointMark.getOffsetLimit() throws when the checkpoint mark was created without an associated KafkaUnboundedReader (the Optional reader is empty). getOffsetLimit() requires a live reader to compute offset-based deduplication limits, so calling it on a reader-less mark is a programming/API-misuse error.","triggerScenarios":"Calling getOffsetLimit() on a KafkaCheckpointMark whose 'reader' Optional is absent, e.g. a mark deserialized from checkpoint storage, a mark constructed manually for testing, or any path outside the normal finalizeCheckpoint/restoring flow.","commonSituations":"Unit tests constructing KafkaCheckpointMark directly; custom runner code inspecting checkpoint marks; calling getOffsetLimit on restored marks before the reader is re-attached.","solutions":["Only call getOffsetLimit() from within a live KafkaUnboundedReader context (reader present).","Check reader.isPresent() before invoking getOffsetLimit().","If you need persisted offsets, read the PartitionMark offsets directly instead of getOffsetLimit().","In tests, construct the mark with a stub KafkaUnboundedReader that reports offsetBasedDeduplicationSupported()."],"exampleFix":"// before\nbyte[] limit = mark.getOffsetLimit();\n// after\nif (mark.reader.isPresent()) {\n  byte[] limit = mark.getOffsetLimit();\n}","handlingStrategy":"type-guard","validationCode":"boolean ok = mark.reader != null && mark.reader.isPresent();","typeGuard":"static boolean readerPresent(KafkaCheckpointMark m){return m!=null&&m.reader!=null&&m.reader.isPresent();}","tryCatchPattern":"try { mark.getOffsetLimit(); } catch (RuntimeException e) { /* fallback to partitions */ }","preventionTips":["Call only within live reader context"],"tags":["kafka","checkpoint","beam-io","state"],"backgroundTag":"null-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}