...

As per the Tiered Storage feature introduced in KIP-405, the  the follower fetch protocol was modified for topics enabled with tiered storage. For such topics, the empty follower finds the offset and leader epoch, up to which the auxiliary state (i.e. leader epoch sequence, producer snapshot state) needs to be built from the leader. The follower then starts fetching the data from the leader starting from that offset. In the KIP, it was decided to use the local-log-start-offset as the offset for this purpose. Instead of the local-log-start-offset, we could also last-tiered-offset which would help the followers quickly catch up with the leader as it only needs to fetch the followers segments that are not moved to tiered storage. The disadvantage 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 locationsstorage, potentially impacting broker performance.

...

In this KIP, we will focus on the follower fetch protocol using the last-tiered-offset.

...

The following diagram provides a visual representation of the leader topic partition's log offsets and will help understand the new follower fetch protocol.

Image Modified

         Lx  = Local log start offset           L_z  = Local log end offset            L_y  = Last stable offset(LSO)

...

    Lz >= Ly >= Lx and Ly >= Ry >= Rx

Let us quickly look at recap the current follower fetch protocol. The follower does the following:

...

We make the following changes to the follower-fetch protocol when the requested offset is not present on the leader (Case 2). These changes apply only when the follower is empty.

• 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 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 builds the remote log auxiliary states for the leader offsets in the range [Log-Start-Offset, Earliest-Pending-Upload-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.

FetchOffet API Changes

We will add a new API for the follower to be able to fetch the pending-upload-offset from the leader. The leader already keeps track of the highest offset that has been uploaded to the remote storage.

• The RLM task on the leader runs at frequent intervals to find new log segments that have become eligible for upload.
• When the task uploads a segment, it also updates the highest offset which has been uploaded to the remote storage.
• If the leader is newly elected, it may not have this information yet and will learn about the highest offset on remote only when a new segment is uploaded.
• We will modify the RLM task behavior to update the highest offset information even if there are no new segments to upload. This ensures that the RLM task has this information ready even it is not able to find new segments eligible for upload.
• The new API returns the offset following the highest offset on the remote as the response to fetch the Earliest-Pending-Upload-Offset

ReplicaFetcher Changes

On receiving an OFFSET_OUT_OF_RANGE or OFFSET_MOVED_TO_TIERED_STORAGE error, ReplicaFetcher does the following:

• Check if the follower replica is empty and if the feature to use last-tiered-offset is enabled. If the check fails, continue with old behaviour.

Follower Fetch Scenarios

...

• 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 same as Leader's Local-Log-Start-Offset.
      • 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 

Whenever this happens (Earliest-Pending-Upload-Offset < Log-Start-Offset), it simply means that they are no valid log segments on the remote yet (otherwise, the leader would return an offset >= log-start-offset). Hence the ReplicaFetcher simply truncates its log and starts replicating all data the leader has locally.

• • Otherwise, ReplicaFetcher builds the remote log auxiliary states for the offset range [Log-Start-Offset, Earliest-Pending-Upload-Offset], sets it Log-End-Offset and Fetch-Offset to Earliest-Pending-Upload-Offset and continues fetching the remaining data from the leader.

Follower Fetch Scenarios

Step 1:

Fetch remote segment info, and rebuild leader epoch sequence.

Step 2:

Continue fetching from the leader

We shall also add a new API for the follower to be able to fetch the pending-upload-offset from the leader. The leader already keeps track of the highest offset that has been uploaded to the remote storage.

• The RLM task on the leader runs at frequent intervals to find new log segments that have become eligible for upload.
• When the task uploads a segment, it also updates the highest offset which has been uploaded to the remote storage.
• If the leader is newly elected, it may not have this information yet and will learn about the highest offset on remote only when a new segment is uploaded.
• We will modify the RLM task behaviour to update the highest offset information even if there are no new segments to upload. This ensures that the RLM task has this information ready even it is not able to find new segments eligible for upload.
• The new API returns the offset following the highest offset on the remote as the response to fetch pending-upload-offset

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?
• If we are changing behavior how will we phase out the older behavior?
• If we need special migration tools, describe them here.
• When will we remove the existing behavior?

Test Plan

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

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.