It is only necessary to override this if {@link #requiresDeduping} has been overridden to\n * return true.\n *\n *
For example, this could be a hash of the record contents, or a logical ID present in the\n * record. If this is generated as a hash of the record contents, it should be at least 16 bytes\n * (128 bits) to avoid collisions.\n *\n *
This method has the same restrictions on when it can be called as {@link #getCurrent} and\n * {@link #getCurrentTimestamp}.\n *\n * @throws NoSuchElementException if the reader is at the beginning of the input and {@link\n * #start} or {@link #advance} wasn't called, or if the last {@link #start} or {@link\n * #advance} returned {@code false}.\n */\n public byte[] getCurrentRecordId() throws NoSuchElementException {\n if (getCurrentSource().requiresDeduping()) {\n throw new IllegalStateException(\n \"getCurrentRecordId() must be overridden if requiresDeduping returns true()\");\n }\n return EMPTY;\n }\n\n /* Returns the offset for the current record of this unbounded reader. */\n public byte[] getCurrentRecordOffset() {\n throw new RuntimeException(\n \"UnboundedReader must override getCurrentRecordOffset() if offset-based deduplication is enabled for the UnboundedSource.\");\n }\n\n /**\n * Returns a timestamp before or at the timestamps of all future elements read by this reader.\n *\n *
This can be approximate. If records are read that violate this guarantee, they will be\n * considered late, which will affect how they will be processed. See {@link\n * org.apache.beam.sdk.transforms.windowing.Window} for more information on late data and how to\n * handle it.","sourceCodeStart":204,"sourceCodeEnd":240,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java#L204-L240","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Override getCurrentRecordId() in your UnboundedReader and return a stable unique id per record (e.g. offset or message id bytes).","If the source cannot produce stable ids, change requiresDeduping() to return false and handle duplicates downstream.","Check that the reader, not just the source, implements the dedup API — the check lives in UnboundedReader."],"exampleFix":"// before\nclass MyReader extends UnboundedReader