apache/beam · error · RuntimeException

Unexpected getOffsetLimit() called while…

Error message

Unexpected getOffsetLimit() called while KafkaUnboundedReader not configured for offset deduplication.

What it means

getOffsetLimit() only supports checkpoint marks produced by a source configured for offset-based deduplication. If the attached KafkaUnboundedReader does not report offsetBasedDeduplicationSupported(), the mark cannot provide an offset limit and the library throws this error to signal unsupported usage.

Solutions

1. Configure the KafkaIO source with offset-based deduplication (e.g. withOffsetDeduplication / consumer dedup config) so the reader supports it.
2. Don't call getOffsetLimit() unless the reader supports offset deduplication — gate on reader.get().offsetBasedDeduplicationSupported().
3. Use the regular checkpoint offsets (PartitionMark) instead of getOffsetLimit for non-dedup sources.

Example fix

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

boolean ok = mark.reader.isPresent() && mark.reader.get().offsetBasedDeduplicationSupported();

static boolean dedupSupported(KafkaCheckpointMark m){return m.reader.isPresent()&&m.reader.get().offsetBasedDeduplicationSupported();}

try { mark.getOffsetLimit(); } catch (RuntimeException e) { /* not dedup-enabled */ }

When it happens

Trigger: Calling getOffsetLimit() when reader is present but KafkaUnboundedReader.offsetBasedDeduplicationSupported() returns false — i.e. the source was not created with offset-based deduplication enabled.

Common situations: Mixing marks/readers from sources with different dedup settings; enabling dedup-dependent logic (e.g. custom dedup runners) on a plain KafkaIO.read(); version mismatch where the runner assumes dedup support the source lacks.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• KafkaCheckpointMark reader is not present while calling…
• : closing producer after unrecoverable error. The work…
• : consumer thread is interrupted
• consumerPollingTimeout should be > 0.
• Couldn't infer Coder from

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

Appendix: source

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

    // 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 {
    private static final long MIN_WATERMARK_MILLIS = BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis();

    private String topic;
    private int partition;

View on GitHub (pinned to 12126d8942)