apache/beam · error · RuntimeException

UnboundedSource must enable offset-based deduplication.

Error message

UnboundedSource must enable offset-based deduplication.

What it means

getCurrentRecordOffset() requires the source to support offset-based deduplication; when offsetBasedDeduplicationSupported() is false it throws a RuntimeException stating the UnboundedSource must enable it. Offsets are the basis for dedup/unique ids in this reader.

Solutions

Enable offset-based deduplication on the Kafka UnboundedSource (use KafkaIO's standard source which implements it)
If extending KafkaUnboundedReader, override offsetBasedDeduplicationSupported() to return true and implement offset semantics
Avoid calling getCurrentRecordOffset() on sources that don't advertise support

Example fix

// before
class MySource extends KafkaUnboundedSource { /* dedup not implemented */ }
// after
@Override
public boolean offsetBasedDeduplicationSupported() { return true; }

Defensive patterns

Strategy: validation

Validate before calling

// check support before calling
if (source.getClass().getName().contains("KafkaUnboundedSource")) {
  // standard KafkaIO source supports offset dedup
}

Type guard

boolean offsetsSupported(UnboundedSource<KafkaRecord<?,?>,?> src) {
  try { src.getClass(); return src instanceof org.apache.beam.sdk.io.kafka.KafkaUnboundedSource; }
  catch (Exception e) { return false; }
}

Try / catch

try {
  byte[] off = reader.getCurrentRecordOffset();
} catch (RuntimeException e) {
  // source does not support offset dedup; use standard KafkaIO source
}

Prevention

Use KafkaIO's built-in UnboundedSource which implements offset-based deduplication
Override offsetBasedDeduplicationSupported() in custom subclasses
Don't call offset APIs on sources that don't advertise support

When it happens

Trigger: Calling getCurrentRecordOffset() on a reader whose source was built without offset-based deduplication support (e.g. source not configured with dedup/offset tracking enabled).

Common situations: Custom runners or Beam plumbing that assume offset reporting (getWatermark/dedup) while the Kafka source was constructed without it — often after swapping in a custom UnboundedSource wrapper.

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

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java:312

  }

  @Override
  public byte[] getCurrentRecordId() throws NoSuchElementException {
    if (!offsetBasedDeduplicationSupported()) {
      // Defer result to super if offset deduplication is not supported.
      return super.getCurrentRecordId();
    }
    if (curRecord == null) {
      throw new NoSuchElementException("KafkaUnboundedReader's curRecord is null.");
    }
    return KafkaIOUtils.OffsetBasedDeduplication.getUniqueId(
        curRecord.getTopic(), curRecord.getPartition(), curRecord.getOffset());
  }

  @Override
  public byte[] getCurrentRecordOffset() throws NoSuchElementException {
    if (!offsetBasedDeduplicationSupported()) {
      throw new RuntimeException("UnboundedSource must enable offset-based deduplication.");
    }
    if (curRecord == null) {
      throw new NoSuchElementException("KafkaUnboundedReader's curRecord is null.");
    }
    return KafkaIOUtils.OffsetBasedDeduplication.encodeOffset(curRecord.getOffset());
  }

  @Override
  public long getSplitBacklogBytes() {
    long backlogBytes = 0;
    for (PartitionState<K, V> p : partitionStates) {
      long pBacklog = p.approxBacklogInBytes();
      if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {
        return UnboundedReader.BACKLOG_UNKNOWN;
      }
      backlogBytes += pBacklog;
    }

View on GitHub (pinned to 12126d8942)