apache/beam · error · RuntimeException

KafkaCheckpointMark reader is not present while calling…

Error message

KafkaCheckpointMark reader is not present while calling getOffsetLimit().

What it means

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.

Solutions

1. Only call getOffsetLimit() from within a live KafkaUnboundedReader context (reader present).
2. Check reader.isPresent() before invoking getOffsetLimit().
3. If you need persisted offsets, read the PartitionMark offsets directly instead of getOffsetLimit().
4. In tests, construct the mark with a stub KafkaUnboundedReader that reports offsetBasedDeduplicationSupported().

Example fix

// before
byte[] limit = mark.getOffsetLimit();
// after
if (mark.reader.isPresent()) {
  byte[] limit = mark.getOffsetLimit();
}

boolean ok = mark.reader != null && mark.reader.isPresent();

static boolean readerPresent(KafkaCheckpointMark m){return m!=null&&m.reader!=null&&m.reader.isPresent();}

try { mark.getOffsetLimit(); } catch (RuntimeException e) { /* fallback to partitions */ }

When it happens

Trigger: 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.

Common situations: Unit tests constructing KafkaCheckpointMark directly; custom runner code inspecting checkpoint marks; calling getOffsetLimit on restored marks before the reader is re-attached.

Related errors

• Unexpected getOffsetLimit() called while…
• Attempting to add message
• : closing producer after unrecoverable error. The work…
• : consumer thread is interrupted
• consumerPollingTimeout should be > 0.

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java:75

  @Override
  public void finalizeCheckpoint() {
    reader.ifPresent(r -> r.finalizeCheckpointMarkAsync(this));
    // Is it ok to commit asynchronously, or should we wait till this (or newer) is committed?
    // Often multiple marks would be finalized at once, since we only need to finalize the latest,
    // it is better to wait a little while. Currently maximum delay is same as KAFKA_POLL_TIMEOUT
    // in the reader (1 second).
  }

  @Override
  public String toString() {
    return "KafkaCheckpointMark{partitions=" + Joiner.on(",").join(partitions) + '}';
  }

  @Override
  public byte[] getOffsetLimit() {
    if (!reader.isPresent()) {
      throw new RuntimeException(
          "KafkaCheckpointMark reader is not present while calling getOffsetLimit().");
    }
    if (!reader.get().offsetBasedDeduplicationSupported()) {
      throw new RuntimeException(
          "Unexpected getOffsetLimit() called while KafkaUnboundedReader not configured for offset deduplication.");
    }

    // KafkaUnboundedSource.split() must produce a 1:1 partition to split ratio.
    checkState(partitions.size() == OFFSET_DEDUP_PARTITIONS_PER_SPLIT);
    PartitionMark partition = partitions.get(/* index= */ 0);
    return KafkaIOUtils.OffsetBasedDeduplication.encodeOffset(partition.getNextOffset());
  }

  /**
   * A tuple to hold topic, partition, and offset that comprise the checkpoint for a single
   * partition.
   */
  public static class PartitionMark implements Serializable {

View on GitHub (pinned to 12126d8942)