docs(streams): add javadoc to RocksDBDualCFRangeIterator

Signed-off-by: Joao Pedro Fonseca Dantas <[email protected]>

Author: fonsdant

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java

@@ -275,6 +275,90 @@ public void close() {
         }
     }
 
+    /**
+     * A range-based iterator for RocksDB that merges results from two column families.
+     *
+     * <p>This iterator supports traversal over two RocksDB column families: one containing timestamped values and
+     * another containing non-timestamped values. It ensures that the keys from both column families are merged and
+     * sorted lexicographically, respecting the iteration order (forward or reverse) and the specified range
+     * boundaries.</p>
+     *
+     * <h2>Key Features</h2>
+     *
+     * <ul>
+     *     <li>Merges results from the "with-timestamp" and "no-timestamp" column families.</li>
+     *     <li>Supports range-based queries with open or closed boundaries.</li>
+     *     <li>Handles both forward and reverse iteration seamlessly.</li>
+     *     <li>Ensures correct handling of inclusive and exclusive upper boundaries.</li>
+     *     <li>Integrates efficiently with Kafka Streams state store mechanisms.</li>
+     * </ul>
+     *
+     * <h2>Usage</h2>
+     *
+     * <p>The iterator can be used for different types of range-based operations, such as:
+     * <ul>
+     *     <li>Iterating over all keys within a range.</li>
+     *     <li>Prefix-based scans (when combined with dynamically calculated range endpoints).</li>
+     *     <li>Open-ended range queries (e.g., from a given key to the end of the dataset).</li>
+     * </ul>
+     * </p>
+     *
+     * <h2>Implementation Details</h2>
+     *
+     * <p>The class extends {@link AbstractIterator} and implements {@link ManagedKeyValueIterator}. It uses RocksDB's
+     * native iterators for efficient traversal of keys within the specified range. Keys from the two column families
+     * are merged during iteration, ensuring proper order and de-duplication where applicable.</p>
+     *
+     * <h3>Key Methods:</h3>
+     *
+     * <ul>
+     *     <li><b>{@code makeNext()}:</b> Retrieves the next key-value pair in the merged range, ensuring
+     *     the result is within the specified range and boundary conditions.</li>
+     *     <li><b>{@code initializeIterators()}:</b> Initializes the RocksDB iterators based on the specified range and direction.</li>
+     *     <li><b>{@code isInRange()}:</b> Verifies if the current key-value pair is within the range defined by {@code from} and {@code to}.</li>
+     *     <li><b>{@code fetchNextKeyValue()}:</b> Determines the next key-value pair to return based on the state of both iterators.</li>
+     * </ul>
+     *
+     * <h3>Thread Safety:</h3>
+     *
+     * <p>The iterator is thread-safe for sequential operations but should not be accessed concurrently from multiple
+     * threads without external synchronization.</p>
+     *
+     * <h2>Examples</h2>
+     *
+     * <h3>Iterate over a range:</h3>
+     *
+     * <pre>{@code
+     * RocksIterator noTimestampIterator = accessor.newIterator(noTimestampColumnFamily);
+     * RocksIterator withTimestampIterator = accessor.newIterator(withTimestampColumnFamily);
+     *
+     * try (RocksDBDualCFRangeIterator iterator = new RocksDBDualCFRangeIterator(
+     *         new Bytes("keyStart".getBytes()),
+     *         new Bytes("keyEnd".getBytes()),
+     *         noTimestampIterator,
+     *         withTimestampIterator,
+     *         "storeName",
+     *         true,  // Forward iteration
+     *         true   // Inclusive upper boundary
+     * )) {
+     *     while (iterator.hasNext()) {
+     *         KeyValue<Bytes, byte[]> entry = iterator.next();
+     *         System.out.println("Key: " + entry.key + ", Value: " + Arrays.toString(entry.value));
+     *     }
+     * }
+     * }</pre>
+     *
+     * <h2>Exceptions</h2>
+     *
+     * <ul>
+     *     <li><b>{@link InvalidStateStoreException}:</b> Thrown if the iterator is accessed after being closed.</li>
+     *     <li><b>{@link IllegalStateException}:</b> Thrown if the close callback is not properly set before usage.</li>
+     * </ul>
+     *
+     * @see AbstractIterator
+     * @see ManagedKeyValueIterator
+     * @see RocksDBStore
+     */
     private static class RocksDBDualCFRangeIterator extends AbstractIterator<KeyValue<Bytes, byte[]>> implements ManagedKeyValueIterator<Bytes, byte[]> {
         private Runnable closeCallback;
         private byte[] noTimestampNext;
@@ -288,6 +372,20 @@ private static class RocksDBDualCFRangeIterator extends AbstractIterator<KeyValu
         private final byte[] rawLastKey;
         private volatile boolean open = true;
 
+        /**
+         * Constructs a new {@code RocksDBDualCFRangeIterator}.
+         *
+         * <p>Initializes the RocksDB iterators for two column families (timestamped and non-timestamped) and sets up
+         * the range and direction for iteration.</p>
+         *
+         * @param from                  The starting key of the range. Can be {@code null} for an open range.
+         * @param to                    The ending key of the range. Can be {@code null} for an open range.
+         * @param noTimestampIterator   The iterator for the non-timestamped column family.
+         * @param withTimestampIterator The iterator for the timestamped column family.
+         * @param storeName             The name of the store associated with this iterator.
+         * @param forward               {@code true} for forward iteration; {@code false} for reverse iteration.
+         * @param toInclusive           Whether the upper boundary of the range is inclusive.
+         */
         RocksDBDualCFRangeIterator(final Bytes from,
                                    final Bytes to,
                                    final RocksIterator noTimestampIterator,
@@ -304,6 +402,15 @@ private static class RocksDBDualCFRangeIterator extends AbstractIterator<KeyValu
             this.rawLastKey = initializeIterators(from, to);
         }
 
+        /**
+         * Retrieves the next key-value pair in the range.
+         *
+         * <p>This method determines the next key-value pair to return by merging the results from the two column
+         * families. If both column families have keys, it selects the one that matches the iteration order and range
+         * conditions. Keys outside the specified range are skipped.</p>
+         *
+         * @return The next {@link KeyValue} pair in the range, or {@code null} if no more elements are available.
+         */
         @Override
         protected KeyValue<Bytes, byte[]> makeNext() {
             loadNextKeys();
@@ -312,24 +419,58 @@ protected KeyValue<Bytes, byte[]> makeNext() {
             return isInRange(next) ? next : allDone();
         }
 
+        /**
+         * Returns the next key in the range without advancing the iterator.
+         *
+         * <p>This method retrieves the key of the next {@link KeyValue} pair that would be returned by {@link #next()},
+         * without moving the iterator forward. This is useful for inspecting the next key without affecting the
+         * iterator's state.</p>
+         *
+         * @return The next key as a {@link Bytes} object.
+         *
+         * @throws NoSuchElementException If there are no more elements in the iterator.
+         */
         @Override
         public Bytes peekNextKey() {
             if (!hasNext()) throw new NoSuchElementException();
             return super.peek().key;
         }
 
+        /**
+         * Advances the iterator and returns the next key-value pair.
+         *
+         * @return The next {@link KeyValue} pair in the range.
+         *
+         * @throws InvalidStateStoreException If the iterator has been closed.
+         */
         @Override
         public synchronized KeyValue<Bytes, byte[]> next() {
             ensureOpen();
             return super.next();
         }
 
+        /**
+         * Checks if there are more elements available in the range.
+         *
+         * @return {@code true} if the iterator has more elements; {@code false} otherwise.
+         *
+         * @throws InvalidStateStoreException If the iterator has been closed.
+         */
         @Override
         public synchronized boolean hasNext() {
             ensureOpen();
             return super.hasNext();
         }
 
+        /**
+         * Closes the iterator and releases associated resources.
+         *
+         * <p>This method ensures that the RocksDB iterators for both column families are properly closed. After this
+         * method is called, any subsequent calls to {@link #hasNext()}, {@link #next()}, or {@link #peekNextKey()} will
+         * result in an {@link InvalidStateStoreException}.</p>
+         *
+         * @throws IllegalStateException If the close callback has not been set before calling this method.
+         */
         @Override
         public synchronized void close() {
             if (closeCallback == null) {
@@ -343,6 +484,11 @@ public synchronized void close() {
             open = false;
         }
 
+        /**
+         * Registers a callback to be executed when the iterator is closed.
+         *
+         * @param closeCallback A {@link Runnable} to execute during the {@link #close()} operation.
+         */
         @Override
         public void onClose(final Runnable closeCallback) {
             this.closeCallback = closeCallback;
@@ -357,6 +503,15 @@ private KeyValue<Bytes, byte[]> compareAndHandleKeys() {
             }
         }
 
+        /**
+         * Determines the next key-value pair to return.
+         *
+         * <p>If one of the column family iterators is exhausted, the method returns the result from the other iterator.
+         * If both iterators have keys, the method compares the keys and returns the appropriate result based on the
+         * iteration direction.</p>
+         *
+         * @return The next {@link KeyValue} pair to return.
+         */
         private KeyValue<Bytes, byte[]> fetchNextKeyValue() {
             if (noTimestampNext == null) {
                 return handleWithTimestampOnly();
@@ -381,6 +536,16 @@ private KeyValue<Bytes, byte[]> handleWithTimestampOnly() {
             return result;
         }
 
+        /**
+         * Checks if the given key-value pair is within the specified range.
+         *
+         * <p>The method compares the key against the range's upper boundary ({@code rawLastKey}) and determines if it
+         * falls within the range.</p>
+         *
+         * @param keyValue The key-value pair to check.
+         *
+         * @return {@code true} if the key is within the range; {@code false} otherwise.
+         */
         private boolean isInRange(final KeyValue<Bytes, byte[]> keyValue) {
             if (rawLastKey == null) return true; // Open-ended range
             final int comparison = comparator.compare(keyValue.key.get(), rawLastKey);
@@ -389,6 +554,17 @@ private boolean isInRange(final KeyValue<Bytes, byte[]> keyValue) {
                     : comparison > 0 || (toInclusive && comparison == 0);
         }
 
+        /**
+         * Initializes the RocksDB iterators based on the specified range and direction.
+         *
+         * <p>This method positions the iterators at the starting point of the range and determines the raw byte
+         * representation of the upper boundary (if provided).</p>
+         *
+         * @param from The starting key of the range. Can be {@code null} for an open range.
+         * @param to   The ending key of the range. Can be {@code null} for an open range.
+         *
+         * @return The raw byte representation of the upper boundary, or {@code null} if no boundary is specified.
+         */
         private byte[] initializeIterators(final Bytes from, final Bytes to) {
             if (forward) {
                 seekIterator(from, withTimestampIterator, true);
@@ -408,11 +584,22 @@ private void ensureOpen() {
             }
         }
 
+        /**
+         * Loads the next keys from the iterators if they are valid.
+         *
+         * <p>This method checks whether the next key for each column family is null. If the corresponding iterator is
+         * valid, it fetches the next key.</p>
+         */
         private void loadNextKeys() {
             if (noTimestampNext == null && noTimestampIterator.isValid()) noTimestampNext = noTimestampIterator.key();
             if (withTimestampNext == null && withTimestampIterator.isValid()) withTimestampNext = withTimestampIterator.key();
         }
 
+        /**
+         * Advances the given iterator based on the iteration direction.
+         *
+         * @param iterator The {@link RocksIterator} to advance.
+         */
         private void moveIterator(final RocksIterator iterator) {
             if (forward) {
                 iterator.next();
@@ -421,6 +608,16 @@ private void moveIterator(final RocksIterator iterator) {
             }
         }
 
+        /**
+         * Seeks the iterator to the specified position.
+         *
+         * <p>If the target is {@code null}, the iterator is positioned at the start or end of the dataset, depending on
+         * the direction.</p>
+         *
+         * @param iterator The {@link RocksIterator} to seek.
+         * @param target   The target key to seek to. Can be {@code null}.
+         * @param forward  {@code true} for forward iteration; {@code false} for reverse.
+         */
         private void seekIterator(final Bytes target, final RocksIterator iterator, final boolean forward) {
             if (target == null) {
                 if (forward) {