[I] [Bug] Log fetch may get stuck when EARLIEST resolves to a lake-only offset [fluss]

["yangshangqing95 (via GitHub)" <gi...@apache.org>]

yangshangqing95 opened a new issue, #4676:
URL: https://github.com/apache/fluss/issues/4676

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
   
   
   ### Fluss version
   
   1.0.0 (latest release)
   
   ### Please describe the bug 🐞
   
   ### Description
   
   Currently, `LogTablet.logStartOffset()` returns the minimum start offset across local log, remote log, and lake storage.
   
   When older log data has expired from local and remote storage but remains available in the lake, `EARLIEST_OFFSET` may resolve to an offset that is no longer directly readable through the Log Fetch API.
   
   In `ReplicaManager.handleFetchOutOfRangeException()`, when the requested offset falls within a lake-only range, the server currently returns an empty successful response without providing records or offset progress. As the existing TODO indicates, fetching directly from the lake is not yet supported through this path.
   
   This can leave the scanner repeatedly fetching an unavailable offset without making progress, even when newer log records are available.
   
   The issue predates bounded scanning and can also affect existing unbounded scans. However, bounded scans make the problem more apparent because they need to reach a specified stopping offset and report completion.
   
   ### Example
   
   Consider the following storage state:
   
   ```text
   Lake-only log range:        [100, 300)
   Local log range:            [300, 400)
   Remote log:                 unavailable for [100, 300)
   High watermark:             400
   Bounded stopping offset:   350
   ```
   
   The earliest offset across all storage tiers is `100`, but the earliest offset directly fetchable through the Log Fetch API is `300`.
   
   **Existing unbounded scan**
   
   ```java
   logScanner.subscribe(bucket, EARLIEST_OFFSET);
   ```
   
   The following sequence can occur:
   
   ```text
   1. FetchLog(EARLIEST_OFFSET)
      -> Server resolves EARLIEST to 100
      -> Offset falls within the lake-only range
      -> Returns empty records without an error or offset progress
   
   2. Client discards the empty response.
      -> Current fetch offset remains EARLIEST_OFFSET (-2)
   
   3. FetchLog(EARLIEST_OFFSET)
      -> Same empty response
   
   4. Subsequent fetches repeat without making progress.
   ```
   
   In this case, the scanner may never reach the newer records starting at offset `300`, even though those records are available in local log storage.
   
   **Bounded scan**
   
   With the bounded scanning support introduced in #4545:
   
   ```java
   logScanner.subscribeBounded(bucket, EARLIEST_OFFSET, 350);
   ```
   
   The behavior is slightly different:
   
   ```text
   1. FetchLog(EARLIEST_OFFSET)
      -> Server resolves EARLIEST to 100
      -> Returns empty records with resolved_earliest_offset = 100
   
   2. Client advances its position to 100.
   
   3. FetchLog(100)
      -> Offset is still within the lake-only range
      -> Returns empty records without offset progress
   
   4. FetchLog(100)
      -> Same empty response
   
   5. The scanner cannot advance toward its stopping offset (350).
   ```
   
   Both paths share the same underlying issue: the earliest offset across all storage tiers does not necessarily represent a position from which the Log Fetch API can make progress.
   
   ### Changelog considerations
   
   The exact behavior also depends on the table type.
   
   Currently, `LogTablet.canFetchFromLakeLog()` returns `false` for changelog tables because direct changelog reads from the lake are not supported.
   
   Therefore, a primary-key table's changelog may receive `LogOffsetOutOfRangeException` instead of the empty successful response described above, depending on remote log availability.
   
   The no-progress loop specifically concerns paths where `canFetchFromLakeLog()` succeeds and the server returns empty records without progress.
   
   The eventual solution should account for these differences rather than assuming identical behavior for all table types.
   
   ### Solution
   
   ### Expected behavior
   
   Log Fetch should not repeatedly return empty responses for an unavailable offset without either making progress or reporting an appropriate error.
   
   We need to clarify the semantics of `EARLIEST_OFFSET` when the earliest data exists only in the lake but cannot be served through the current Log Fetch API.
   
   Possible directions include:
   
   - Distinguishing the earliest offset across all storage tiers from the earliest offset directly fetchable through the Log Fetch API.
   - Providing a meaningful next offset or another explicit progress indication when a fetch cannot return records from a lake-only range.
   - Reporting an appropriate error when the requested data is unavailable through Log Fetch and advancing the offset would silently skip required records.
   - Supporting direct reads from lake storage in the future, if appropriate.
   
   The solution must preserve data consistency. In particular, automatically advancing to the earliest locally or remotely fetchable offset should not silently skip data that the caller expects to consume.
   
   
   ### Are you willing to submit a PR?
   
   - [x] I'm willing to submit a PR!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org