{"record":{"id":"dd0794cc1853837a","repo":"apache/beam","slug":"kafkaunboundedreader-s-currecord-is-null","errorCode":null,"errorMessage":"KafkaUnboundedReader's curRecord is null.","messagePattern":"KafkaUnboundedReader's curRecord is null\\.","errorType":"exception","errorClass":"NoSuchElementException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java","lineNumber":303,"sourceCode":"    return curRecord;\n  }\n\n  @Override\n  public Instant getCurrentTimestamp() throws NoSuchElementException {\n    if (curTimestamp == null) {\n      throw new NoSuchElementException();\n    }\n    return curTimestamp;\n  }\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() {","sourceCodeStart":285,"sourceCodeEnd":321,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java#L285-L321","documentation":"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.","triggerScenarios":"Calling getCurrentRecordId() while offset-based deduplication is enabled but no record is current (curRecord == null).","commonSituations":"Runner deduplication plumbing querying the record id before the first record is read, or after an empty poll.","solutions":["Only query the record id when a record is current","If you don't need dedup ids, don't enable offset-based deduplication on the source","Check reader lifecycle: ensure start()/advance() populated a record first"],"exampleFix":null,"handlingStrategy":"type-guard","validationCode":null,"typeGuard":"boolean canGetRecordId(KafkaUnboundedReader<?,?> r) {\n  try { r.getCurrentRecordId(); return true; }\n  catch (NoSuchElementException e) { return false; }\n}","tryCatchPattern":"try {\n  byte[] id = reader.getCurrentRecordId();\n} catch (NoSuchElementException e) {\n  // curRecord null: no record to dedupe yet\n}","preventionTips":["Query record ids only during active record processing","Enable offset-based deduplication if dedup ids are required"],"tags":["java","kafka","deduplication","nosuchelement"],"backgroundTag":"empty-result-set","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"}