...

If the target timestamp is

ListOffsetRequest.EARLIEST_TIMESTAMP (value as -2) returns logStartOffset of the log.

ListOffsetRequest.LATEST_TIMESTAMP(value as -1) returns log-stable-offset or log-end-offset based on the isolation level in the request.

ListOffsetRequest.MAX_TIMESTAMP (value as -3) returns the offset corresponding to the record with the highest timestamp on the partition.

ListOffsetRequest.EARLIEST_LOCAL_TIMESTAMP (value as -4) returns the earliest offset stored in the local log.

ListOffsetRequest.LAST_TIERED_TIMESTAMP (value as -5) returns the latest tiered offset.

This API will be enhanced with supporting new target timestamp with a value of -6 which is called EARLIEST_PENDING_UPLOAD_OFFSET_TIMESTAMP. There will not be any new fields added in the request and response schemas but there will be a version bump to indicate the version update. This offset represents that offset on the leader's local log storage that is the next one to be uploaded to the remote storage (all the previous offsets have already been tiered).

Proposed Changes

High-Level Design

...

• Leader responds with either OFFSET_OUT_OF_RANGE or OFFSET_MOVED_TO_TIERED_STORAGE error
• Follower fetches the earliest offset on the leader (Log-Start-Offset), if not available in the leader response
• Follower fetches the earliest offset on the leader that is waiting to be uploaded (Earliest-Pending-Upload-Offset). This is the next offset after the last-tiered-offset.
• The follower needs the offset and the leader epoch for the offset following the last-tiered-offset, i.e. last-tiered-offset + 1
Instead of fetching
• Follower builds the remote log auxiliary states for the leader offsets in the range [Log-Start-Offset, Earliest-Pending-Upload-Offset, it could fetch the last-tiered-offset from the leader, and make a separate leader call to fetch leader epoch for the following offset.)
• Follower updates its offsets
  • Log-Start-Offset → Leader's Log-Start-Offset
  • Log-End-Offset →
  Follower builds the remote log auxiliary states for the leader offsets in the range [Log-Start-Offset,
  • Earliest-Pending-Upload-Offset
  )
  • Fetch-Offset
  Follower updates its offsets
  • Log-Start-Offset → Leader's Log-Start-Offset
  • Log-End-Offset → Earliest-Pending-Upload-Offset
  • Fetch-Offset → Earliest-Pending-Upload-Offset
• The follower requests data from the new fetch offset.

...

Unit Tests and Integration Tests for the various follower fetch scenarios.

Rejected Alternatives

The empty follower needs the offset (and the corresponding leader epoch) that is the next in line to be uploaded to the remote storage. As discussed above, we will introduce a new target timestamp in ListOffsetsRequest API to fetch this offset and the leader epoch from the leader.

Another way to get the same offset and leader epoch is to use the existing target timestamp supported by ListOffsetsRequest API. For non-compacted tiered-storage enabled topics, EARLIEST-PENDING-UPLOAD-OFFSET is LAST-TIERED-OFFSET + 1. The follower could fetch the LAST-TIERED-OFFSET from the leader and add 1 to it to get the EARLIEST-PENDING-UPLOAD-OFFSET. However, it will also need the leader epoch for EARLIEST-PENDING-UPLOAD-OFFSET and needs to make another call to the leader to fetch the corresponding leader epoch. This alternative was rejected because of the following reasons:

1. It is more complicated than the proposed approach and requires multiple calls to the leader.
2. In the future, if we support compacted topics on tiered storage, the logic of adding 1 to LAST-TIERED-OFFSET may not work because the offsets are not guaranteed to be contiguous. The follower protocol will need significant changes to handle this. Keeping the logic to discover EARLIEST-PENDING-UPLOAD-OFFSET on the leader instead will make it simpler to handle compacted topics in the future.