{"record":{"id":"d5a3d7af6cc87a7f","repo":"apache/beam","slug":"unboundedsource-must-enable-offset-based-deduplication","errorCode":null,"errorMessage":"UnboundedSource must enable offset-based deduplication.","messagePattern":"UnboundedSource must enable offset-based deduplication\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java","lineNumber":312,"sourceCode":"  }\n\n  @Override\n  public byte[] getCurrentRecordId() throws NoSuchElementException {\n    if (!offsetBasedDeduplicationSupported()) {\n      // Defer result to super if offset deduplication is not supported.\n      return super.getCurrentRecordId();\n    }\n    if (curRecord == null) {\n      throw new NoSuchElementException(\"KafkaUnboundedReader's curRecord is null.\");\n    }\n    return KafkaIOUtils.OffsetBasedDeduplication.getUniqueId(\n        curRecord.getTopic(), curRecord.getPartition(), curRecord.getOffset());\n  }\n\n  @Override\n  public byte[] getCurrentRecordOffset() throws NoSuchElementException {\n    if (!offsetBasedDeduplicationSupported()) {\n      throw new RuntimeException(\"UnboundedSource must enable offset-based deduplication.\");\n    }\n    if (curRecord == null) {\n      throw new NoSuchElementException(\"KafkaUnboundedReader's curRecord is null.\");\n    }\n    return KafkaIOUtils.OffsetBasedDeduplication.encodeOffset(curRecord.getOffset());\n  }\n\n  @Override\n  public long getSplitBacklogBytes() {\n    long backlogBytes = 0;\n    for (PartitionState<K, V> p : partitionStates) {\n      long pBacklog = p.approxBacklogInBytes();\n      if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {\n        return UnboundedReader.BACKLOG_UNKNOWN;\n      }\n      backlogBytes += pBacklog;\n    }\n","sourceCodeStart":294,"sourceCodeEnd":330,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java#L294-L330","documentation":"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.","triggerScenarios":"Calling getCurrentRecordOffset() on a reader whose source was built without offset-based deduplication support (e.g. source not configured with dedup/offset tracking enabled).","commonSituations":"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.","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"],"exampleFix":"// before\nclass MySource extends KafkaUnboundedSource { /* dedup not implemented */ }\n// after\n@Override\npublic boolean offsetBasedDeduplicationSupported() { return true; }","handlingStrategy":"validation","validationCode":"// check support before calling\nif (source.getClass().getName().contains(\"KafkaUnboundedSource\")) {\n  // standard KafkaIO source supports offset dedup\n}","typeGuard":"boolean offsetsSupported(UnboundedSource<KafkaRecord<?,?>,?> src) {\n  try { src.getClass(); return src instanceof org.apache.beam.sdk.io.kafka.KafkaUnboundedSource; }\n  catch (Exception e) { return false; }\n}","tryCatchPattern":"try {\n  byte[] off = reader.getCurrentRecordOffset();\n} catch (RuntimeException e) {\n  // source does not support offset dedup; use standard KafkaIO source\n}","preventionTips":["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"],"tags":["java","kafka","deduplication","unsupported-operation"],"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"}