...

A drawback of using the last-tiered-offset is that this new follower would possess only a limited number of locally stored segments. Should it ascend to the role of leader, there is a risk of needing to fetch these segments from the remote storage, potentially impacting broker performance. There can be several ways to prevent such followers from transitioning to a leader role, such as having strict leader election criteria, wherein followers with limited locally stored segments are not allowed to become a leader, or having slightly relaxed criteria wherein other followers are preferred for leadership over this follower. A similar problem can also happen on clusters that are enabled with fetch from the closest replica (KIP-392). For such clusters, the follower that was bootstrapped using the tiered offset may need to serve fetch requests. To prevent the degradation of the follower, we can build a similar mechanism to exclude/deprioritize them for consumption requests. Incorporating these preventing measures alongside the proposed utilization of last-tiered-offset can effectively balance the advantage of rapid follower synchronization and mitigate potential performance impacts on the broker, thereby optimizing the overall efficiency and reliability.

...

• Type: Boolean
• Mode: Dynamically configurable as cluster-default for all brokers in the cluster.
• Description: Whether the last tiered offset should be used as the start offset for bootstrapping an empty follower
• Default value: False

List Offsets API

ListOffsets API gives the offset(s) for the given timestamp either by looking into the local log or remote log time indexes.

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

...

• Check if the follower replica is empty and if the feature to use last-tiered-offset is enabled. If the check fails, continue with the old behaviour.
• Otherwise, do the following:
  • Fetch the leader's log-start-offset
  • Fetch the leader's Earliest-Pending-Upload-Offset using the API discussed in the previous section
  • If the Earliest-Pending-Upload-Offset is unknown (API returned -1), there are two possible cases:
    • No segments for the partition have been uploaded to the remote storage yet
      • It can be confirmed by checking if the leader's Log-Start-Offset is the same as the Leader's Local-Log-Start-Offset.
        • The follower will make another call to the leader to fetch the EarliestLocal offset (same as Leader's Local-Log-Start-Offset) to be able to verify this scenario.
      • If confirmed, we use the Local-Log-Start-Offset as the Earliest-Pending-Upload-Offset
    • Segments are uploaded to the remote storage but the leader does not know yet.
      • In this case, the leader will eventually learn about the Earliest-Pending-Upload-Offset.
      • The ReplicaFetcher should throw an exception so that the partition can be retried after some time (while the leader learns about the offset)
  • The highest offset on the remote storage may be much lower than the leader's Log-Start-Offset. Hence, the Earliest-Pending-Upload-Offset returned by the leader may also be less than the leader's Log-Start-Offset. This may happen because of the following situation:
    • Tiered Storage was enabled for the topics and log segments were uploaded to the remote storage
    • Later, tiered storage was disabled for the topic and more messages were produced on the topic
    • Some time has elapsed and now the leader's Log-Start-Offset is much ahead of the highest offset on the remote storage.
    • Now tiered storage is enabled for the topic, but no segments have been uploaded yet because no new segments are eligible to be uploaded yet 

...

Fetch remote segment info, and rebuild leader epoch sequence.

Step 2:

Continue fetching from the leader

Upgrade

The feature will be guarded by a new metadata version and will not be enabled by default during the rolling upgrade. Follow the steps mentioned in Kafka upgrade to reach the state where all brokers are running on the latest binaries with the new "inter.broker.protocol" version.

Test Plan

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