This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Draft

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

As per the Tiered Storage feature introduced in KIP-405, the follower fetch protocol was modified for topics enabled with tiered storage. For such topics, the 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 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 remote locations, potentially impacting broker performance.

There can be multiple 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. 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.

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

Public Interfaces

Broker Configuration options

A new broker property will be added to enable/disable the feature of using last-tiered-offset in the follower fetch.

follower.fetch.last.tiered.offset.enable

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

Proposed Changes

High-Level Design

The following diagram provides a visual representation of the leader topic partiton's log offsets.

Screenshot 2019-10-25 at 7.14.08 PM.png

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

R_y = Remote log end offset / Last Tiered Offset R_x = Remote log start offset

L_z >= L_y>= Lx and L_y >= R_y>= R_x

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

It requests data from the leader, beginning from its Fetch-Offset
- If the follower is empty, the Fetch-Offset is zero.
The requested offset may or may not be a valid offset on the leader
- Case 1: The requested offset is present on the leader
  - Leader responds with the data to the follower
  - Follower receives the data, appends records to the local disk, updates its Log-End-Offset and its Fetch-Offset
  - Follower requests data beginning from the new fetch offset
- Case 2: The requested offset is not present on the leader. There are two possible scenarios here:
  - Case 2.1: The requested offset is lower than the earliest offset the leader knows
    - Leader response with OFFSET_OUT_OF_RANGE error
    - Follower fetches the earliest offset on the leader (Log-Start-Offset)
    - Follower updates its offsets
      - Log-Start-Offset → leader's Log-Start-Offset
      - Log-End-Offset → leader's Log-Start-Offset
      - Fetch-Offset → leader's Log-Start-Offset
    - Follower requests data from the new fetch offset.
  - Case 2.2: The requested offset is greater than or equal to the earliest offset the leader knows, but the offset has moved to tiered storage.
    - Leader responds with OFFSET_MOVED_TO_TIERED_STORAGE error (also returns its Log-Start-Offset)
    - Follower fetches the leader's Local-Log-Start-Offset
    - Follower builds the remote log auxiliary states for the leader offsets in the range [Log-Start-Offset, Local-Log-Start-Offset]
    - Follower updates its offsets
      - Log-Start-Offset → Leader's Log-Start-Offset
      - Log-End-Offset → Leader's Local-Log-Start-Offset
      - Fetch-Offset → Leader's Local-Log-Start-Offset
    - Follower requests data from the new fetch offset.

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

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.
Follwer 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
Follower requests data from the new fetch offset.

Follower Fetch Scenarios

Scenario 1: New Empty Follower

Step 1:

Fetch remote segment info, and rebuild leader epoch sequence.

Broker A (Leader)

Broker B (Follower)

Remote Storage

RL Metadata Storage

3: msg 3 LE-1

4: msg 4 LE-1

5: msg 5 LE-2

6: msg 6 LE-2

7: msg 7 LE-3 (HW)

leader_epochs:

LE-0, 0

LE-1, 3

LE-2, 5

LE-3, 7

Fetch LE-1, 0
Receives OMTS
Receives ELO 3, LE-1
Fetches EPUO
Receives EPUO 6, LE-2
Fetch remote segment info and build local leader epoch sequence until ELO

leader_epochs:

LE-0,0

LE-1,3

LE-2,5

seg-0-2, uuid-1

log:

0: msg 0 LE-0

1: msg 1 LE-0

2: msg 2 LE-0

epochs:

LE-0, 0

seg 3-5, uuid-2

log:

3: msg 3 LE-1

4: msg 4 LE-1

5: msg 5 LE-2

epochs:

LE-0, 0

LE-1, 3

LE-2, 5

seg-0-2, uuid-1

segment epochs:

LE-0, 0

seg-3-5, uuid-2

segment epochs:

LE-1, 3

LE-2, 5

Step 2:

Continue fetching from the leader

Broker A (Leader)

Broker B (Follower)

Remote Storage

RL Metadata Storage

3: msg 3 LE-1

4: msg 4 LE-1

5: msg 5 LE-2

6: msg 6 LE-2

7: msg 7 LE-3 (HW)

leader_epochs:

LE-0, 0

LE-1, 3

LE-2, 5

LE-3, 7

Fetch from EPUO to HW

6: msg 6 LE-2

7: msg 7 LE-3 (HW)

leader_epochs:

LE-0,0

LE-1,3

LE-2,5

LE-3,7

seg-0-2, uuid-1

log:

0: msg 0 LE-0

1: msg 1 LE-0

2: msg 2 LE-0

epochs:

LE-0, 0

seg 3-5, uuid-2

log:

3: msg 3 LE-1

4: msg 4 LE-1

5: msg 5 LE-2

epochs:

LE-0, 0

LE-1, 3

LE-2, 5

seg-0-2, uuid-1

segment epochs:

LE-0, 0

seg-3-5, uuid-2

segment epochs:

LE-1, 3

LE-2, 5

Scenario 2: Out-of-sync follower catching up

An out-of-sync follower (Broker B) has local data up to offset 3

Step 1:

Broker A (Leader)

Broker B (Follower)

Remote Storage

RL Metadata Storage

0: msg 0 LE-0

1: msg 1 LE-0

2: msg 2 LE-0

3: msg 3 LE-1

4: msg 4 LE-1

5: msg 5 LE-2

6: msg 6 LE-2

7: msg 7 LE-3

8: msg 8 LE-3

9: msg 9 LE-3 (HW)

leader_epochs:

LE-0, 0

LE-1, 3

LE-2, 5

LE-3, 7

0: msg 0 LE-0

1: msg 1 LE-0

2: msg 2 LE-0

3: msg 3 LE-1

leader_epochs:

LE-0, 0

LE-1, 3

1. Because the latest leader epoch in the local storage (LE-1) does not equal the current leader epoch (LE-3). The follower starts from the Truncating state.

2. fetchLeaderEpochEndOffsets(LE-1) returns 5, which is larger than the latest local offset. With the existing truncation logic, the local log is not truncated and it moves to Fetching state.

seg-0-2, uuid-1

log:

0: msg 0 LE-0

1: msg 1 LE-0

2: msg 2 LE-0

epochs:

LE-0, 0

seg 3-5, uuid-2

log:

3: msg 3 LE-1

4: msg 4 LE-1

5: msg 5 LE-2

epochs:

LE-0, 0

LE-1, 3

LE-2, 5

seg-0-2, uuid-1

segment epochs:

LE-0, 0

seg-3-5, uuid-2

segment epochs:

LE-1, 3

LE-2, 5

Step 2:

Broker A (Leader)

Broker B (Follower)

Remote Storage

RL Metadata Storage

9: msg 9 LE-3

10: msg 10 LE-3

11: msg 11 LE-3 (HW)

[segments till offset 8 were deleted]

leader_epochs

LE-0, 0

LE-1, 3

LE-2, 5

LE-3, 7

0: msg 0 LE-0

1: msg 1 LE-0

2: msg 2 LE-0

3: msg 3 LE-1

leader_epochs:

LE-0, 0

LE-1, 3

1. Fetch from leader LE-1, 4

2. Receives OMTS, truncates local segments.

3. Fetch ELO, Receives ELO 9, LE-3, and moves to BuildingRemoteLogAux state

seg-0-2, uuid-1

log:

0: msg 0 LE-0

1: msg 1 LE-0

2: msg 2 LE-0

epochs:

LE-0, 0

seg 3-5, uuid-2

log:

3: msg 3 LE-1

4: msg 4 LE-1

5: msg 5 LE-2

epochs:

LE-0, 0

LE-1, 3

LE-2, 5

Seg 6-8, uuid-3, LE-3

log:

6: msg 6 LE-2

7: msg 7 LE-3

8: msg 8 LE-3

epochs:

LE-0, 0

LE-1, 3

LE-2, 5

LE-3, 7

seg-0-2, uuid-1

segment epochs:

LE-0, 0

seg-3-5, uuid-2

segment epochs:

LE-1, 3

LE-2, 5

seg-6-8, uuid-3

segment epochs:

LE-2, 5

LE-3, 7

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

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

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.