apache/beam · error · NoSuchElementException

KafkaUnboundedReader's curRecord is null.

Error message

KafkaUnboundedReader's curRecord is null.

What it means

getCurrentRecordId() provides a deduplication id derived from topic/partition/offset. If offset-based deduplication is unsupported it defers to super; otherwise, if curRecord is null it throws NoSuchElementException with an explicit message that the reader has no current record.

Solutions

1. Only query the record id when a record is current
2. If you don't need dedup ids, don't enable offset-based deduplication on the source
3. Check reader lifecycle: ensure start()/advance() populated a record first

boolean canGetRecordId(KafkaUnboundedReader<?,?> r) {
  try { r.getCurrentRecordId(); return true; }
  catch (NoSuchElementException e) { return false; }
}

try {
  byte[] id = reader.getCurrentRecordId();
} catch (NoSuchElementException e) {
  // curRecord null: no record to dedupe yet
}

When it happens

Trigger: Calling getCurrentRecordId() while offset-based deduplication is enabled but no record is current (curRecord == null).

Common situations: Runner deduplication plumbing querying the record id before the first record is read, or after an empty poll.

Understand the failure class

Background: EmptyResultError / "no results found": when an API or scraper succeeds but returns zero rows — this error's family across 9 libraries.

Related errors

• UnboundedSource must enable offset-based deduplication.
• Could not access keytab file. Make sure that the…
• Could not find class:
• Couldn't resolve coder for Deserializer:
• Error matching values. Secret was discovered but its value…

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

Appendix: source

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

    return curRecord;
  }

  @Override
  public Instant getCurrentTimestamp() throws NoSuchElementException {
    if (curTimestamp == null) {
      throw new NoSuchElementException();
    }
    return curTimestamp;
  }

  @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() {

View on GitHub (pinned to 12126d8942)