apache/beam · error · IllegalStateException

getCurrentRecordId() must be overridden if requiresDeduping

Error message

getCurrentRecordId() must be overridden if requiresDeduping returns true()

What it means

UnboundedReader.getCurrentRecordId() returns a stable unique id for the current record so runners can deduplicate records from unbounded sources. The default implementation throws IllegalStateException if the source declares requiresDeduping() == true, because a deduping source must supply real record ids; otherwise it returns an empty byte array.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java:222

     * of the same logical record read from the underlying data source.
     *
     * <p>It is only necessary to override this if {@link #requiresDeduping} has been overridden to
     * return true.
     *
     * <p>For example, this could be a hash of the record contents, or a logical ID present in the
     * record. If this is generated as a hash of the record contents, it should be at least 16 bytes
     * (128 bits) to avoid collisions.
     *
     * <p>This method has the same restrictions on when it can be called as {@link #getCurrent} and
     * {@link #getCurrentTimestamp}.
     *
     * @throws NoSuchElementException if the reader is at the beginning of the input and {@link
     *     #start} or {@link #advance} wasn't called, or if the last {@link #start} or {@link
     *     #advance} returned {@code false}.
     */
    public byte[] getCurrentRecordId() throws NoSuchElementException {
      if (getCurrentSource().requiresDeduping()) {
        throw new IllegalStateException(
            "getCurrentRecordId() must be overridden if requiresDeduping returns true()");
      }
      return EMPTY;
    }

    /* Returns the offset for the current record of this unbounded reader. */
    public byte[] getCurrentRecordOffset() {
      throw new RuntimeException(
          "UnboundedReader must override getCurrentRecordOffset() if offset-based deduplication is enabled for the UnboundedSource.");
    }

    /**
     * Returns a timestamp before or at the timestamps of all future elements read by this reader.
     *
     * <p>This can be approximate. If records are read that violate this guarantee, they will be
     * considered late, which will affect how they will be processed. See {@link
     * org.apache.beam.sdk.transforms.windowing.Window} for more information on late data and how to
     * handle it.

View on GitHub (pinned to 12126d8942)

Solutions

1. Override getCurrentRecordId() in your UnboundedReader and return a stable unique id per record (e.g. offset or message id bytes).
2. If the source cannot produce stable ids, change requiresDeduping() to return false and handle duplicates downstream.
3. Check that the reader, not just the source, implements the dedup API — the check lives in UnboundedReader.

Example fix

// before
class MyReader extends UnboundedReader<MyRecord> {
  // no getCurrentRecordId override
}

// after
class MyReader extends UnboundedReader<MyRecord> {
  @Override
  public byte[] getCurrentRecordId() {
    return Longs.toByteArray(currentRecord.getOffset());
  }
}

if (source.requiresDeduping()) {
  checkState(!(reader.getClass().getMethod("getCurrentRecordId").getDeclaringClass() == UnboundedReader.class), "reader must override getCurrentRecordId()");
}

try {
  byte[] id = reader.getCurrentRecordId();
} catch (IllegalStateException e) {
  // reader does not support dedup ids: fall back to non-dedup processing
}

When it happens

Trigger: Calling getCurrentRecordId() on a reader whose UnboundedSource.requiresDeduping() returns true while the reader does not override getCurrentRecordId() — typically invoked by the runner (processElement/recordId paths) when offset- or id-based dedup is enabled.

Common situations: Writing a custom UnboundedReader for a dedup-required source (e.g. queues with possible redeliveries) and forgetting to implement record ids; upgrading Beam so a runner now asks for record ids on sources that previously returned true from requiresDeduping() without consequence.

Related errors

• CheckpointMark must override getOffsetLimit() if offset-base
• UnboundedReader must override getCurrentRecordOffset() if of
• Failed closing channel to %s
• Error determining if %s allows dynamic splitting
• File spec %s not found

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