{"record":{"id":"5c38acc7665da786","repo":"apache/beam","slug":"unexpected-getoffsetlimit-called-while-kafkaunboundedreader","errorCode":null,"errorMessage":"Unexpected getOffsetLimit() called while KafkaUnboundedReader not configured for offset deduplication.","messagePattern":"Unexpected getOffsetLimit\\(\\) called while KafkaUnboundedReader not configured for offset deduplication\\.","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":79,"sourceCode":"    // 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 {\n    private static final long MIN_WATERMARK_MILLIS = BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis();\n\n    private String topic;\n    private int partition;","sourceCodeStart":61,"sourceCodeEnd":97,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java#L61-L97","documentation":"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.","triggerScenarios":"Calling getOffsetLimit() when reader is present but KafkaUnboundedReader.offsetBasedDeduplicationSupported() returns false — i.e. the source was not created with offset-based deduplication enabled.","commonSituations":"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.","solutions":["Configure the KafkaIO source with offset-based deduplication (e.g. withOffsetDeduplication / consumer dedup config) so the reader supports it.","Don't call getOffsetLimit() unless the reader supports offset deduplication — gate on reader.get().offsetBasedDeduplicationSupported().","Use the regular checkpoint offsets (PartitionMark) instead of getOffsetLimit for non-dedup sources."],"exampleFix":"// before\nbyte[] limit = mark.getOffsetLimit();\n// after\nif (mark.reader.isPresent() && mark.reader.get().offsetBasedDeduplicationSupported()) {\n  byte[] limit = mark.getOffsetLimit();\n}","handlingStrategy":"type-guard","validationCode":"boolean ok = mark.reader.isPresent() && mark.reader.get().offsetBasedDeduplicationSupported();","typeGuard":"static boolean dedupSupported(KafkaCheckpointMark m){return m.reader.isPresent()&&m.reader.get().offsetBasedDeduplicationSupported();}","tryCatchPattern":"try { mark.getOffsetLimit(); } catch (RuntimeException e) { /* not dedup-enabled */ }","preventionTips":["Enable offset dedup on the source"],"tags":["kafka","checkpoint","deduplication","beam-io"],"backgroundTag":"unsupported-operation","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"}