[KAFKA-14685] Refactor logic to handle OFFSET_MOVED_TO_TIERED_STORAGE error by mattwong949 · Pull Request #13206 · apache/kafka

mattwong949 · 2023-02-06T22:08:45Z

This PR adds the TierStateMachine interface to handle all state transitions related to tiered storage and building the remote log aux state.

The new interface supports a start and maybeAdvanceState. In the ReplicaFetcherTierStateMachine, the maybeAdvanceState is unused since the implementation is synchronous. Only the start is needed. This PR keeps the addition of the maybeAdvanceState because there is an existing task for building the remote log aux state in an asynchronous manner that will be able to use the full interface https://issues.apache.org/jira/browse/KAFKA-13560

…also implemented in java. need to modify testing still

…s wip

junrao · 2023-02-06T22:42:39Z

cc @satishd

rittikaadhikari

Thanks for the PR @mattwong949! Few comments.

Have you considered refactoring out some of the tiering specific tests in AbstractFetcherThreadTest into a separate test suite? (i.e., ReplicaFetcherTierStateMachineTest.java)?

satishd

Thanks @mattwong949 for the PR. I had an initial review, and left a few minor comments.

…factor

Hangleton · 2023-02-08T12:23:01Z

+
+            Optional<RemoteLogSegmentMetadata> maybeRlsm = rlm.fetchRemoteLogSegmentMetadata(topicPartition, targetEpoch, previousOffsetToLeaderLocalLogStartOffset);
+
+            if (maybeRlsm.isPresent()) {


Note: if the rlmMetadata is unavailable for an extended period of time, the replica fetcher will keep retrying indefinitely to construct the starting fetch state for the partition. This will lead to an OffsetForLeaderEpoch and ListOffsets requests every time. If a large number of partitions are impacted, that will generate unnecessary inter-broker traffic on the cluster - although marginal most of the time. As an optimization, we could store the leader epoch associated to the leader's local log start offset - 1 which was retrieved here (we would still, however, need to query for the local log start offset on the leader on every iteration, and fetch the associated leader epoch if it has changed).

The asynchronous resolution of the correct fetch state from the remote storages (KAFKA-13560) will prevent the extra load on the replica fetcher thread itself. The consideration above applies to the RPCs which are made on the synchronous fetch path.

Hangleton · 2023-02-08T12:56:09Z

+     * It tries to build the required state for this partition from leader and remote storage so that it can start
+     * fetching records from the leader.
+     */
+    private Long buildRemoteLogAuxState(TopicPartition topicPartition,


Have you considered refactoring this method to ease testability? This could be part of another PR too to avoid changing too many parts in one code refactor.

junrao

@mattwong949 : Thanks for the PR. A few comments below.

junrao · 2023-02-08T19:24:46Z


+    def buildRemoteLog(topicPartition: TopicPartition, leaderLogStartOffset: Long): Unit = {
+      fetcher.truncateFullyAndStartAt(topicPartition, leaderState.localLogStartOffset)
+      replicaState.logStartOffset = leaderLogStartOffset


Here, we want to update the replicaState in the MockFetcherThread. Would it be better to call fetcher.replicaPartitionState to get the replicaState explicitly?

Also, adding a callback in a Mock class seems a bit complicated. Alternatively, we could override doWork() in MockFetcherThread so that it updates its replicaPartitionState from the partitionState in AbstractFetcherThread after each call. This way, we could get rid of the callback in mockTierStateMachine.

I had tried to ensure that the MockTierStateMachine code would get invoked as part of the test as a sanity check that the TierStateMachine logic is getting called from handleOffsetsMovedToTieredStorage. I couldn't think of another way to get the replicaPartitionState updated from the MockTierStateMachine since it is contained in the MockFetcherThread.

I am not sure about the effectiveness of the test if we override the doWork from the MockFetcherThread to update the replicaPartitionState given my thoughts on trying to invoke the MockTierStateMachine code, but maybe I'm misunderstanding the alternative you mentioned.

Well, buildRemoteLog() does two things (1) call fetcher.truncateFullyAndStartAt and (2) set replicaState.logStartOffset. For (1), since the truncation logic is moved to TierStateMachine, it probably should be done in MockTierStateMachine.start() directly. For (2), the existing test doesn't need to set replicaState.logStartOffset. So, it seems it's unnecessary? If address both (1) and (2), then the callback is not needed.

hm ok I think I see what you're suggesting. Let me try the refactor w/o the callback. thanks Jun

showuon

@mattwong949 , made a 1st pass, left some comments. Thanks for the PR.

showuon · 2023-02-10T07:39:40Z

-                  debug(s"Received error ${Errors.OFFSET_MOVED_TO_TIERED_STORAGE}, " +
-                    s"at fetch offset: ${currentFetchState.fetchOffset}, " + s"topic-partition: $topicPartition")
-                  if (!handleOffsetsMovedToTieredStorage(topicPartition, currentFetchState,
-                    fetchPartitionData.currentLeaderEpoch, partitionData.logStartOffset()))


For the logStartOffset, we used to retrieve from partitionData, which is from fetchResponse (from leader), and now, we changed to get from fetchPartitionData, which is from fetchRequest. Any reason we change it? I'm thinking the logStartOffset should still rely on the leader response to avoid some inconsistency. WDYT?

ah i believe this was a mistake. I wanted the TierStateMachine.start to use the fetch response as well. thanks for the catch

showuon · 2023-02-10T07:59:56Z

-      val (epoch, leaderStartOffset) = if (fetchFromLocalLogStartOffset)
-        leader.fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch) else
-        leader.fetchEarliestOffset(topicPartition, currentLeaderEpoch)
-
+      val (_, leaderStartOffset) = leader.fetchEarliestOffset(topicPartition, currentLeaderEpoch)


Could you explain why we change this behavior? I think we should try fetch from local start offset if possible to save time to catch up. But after this change, we always fetch from log start offset (not local log start offset). Why should we change it?

IIUC, this is correct. We should try to fetch from the leader log start offset instead of the local leader log start offset so that in the case where the leader log start offset < local leader log start offset, the leader returns an offset-move-to-tiered-storage error and the follower takes the related code path to reconstruct the local replica log prefix.

+1 to @Hangleton 's comment. I restored this function back to the original logic, so this function is only called when first trying to fetch the leader log start offset (after starting fetch or after getting the offset out of range error). If the follower gets the OFFSET_MOVED_TO_TIERED_STORAGE error, it proceeds to other the code path to build the remote aux log state via TierStateMachine interface

showuon · 2023-02-10T08:03:16Z

+       * In the first case, the follower's current log end offset is smaller than the leader's log start offset. So the
+       * follower should truncate all its logs, roll out a new segment and start to fetch from the current leader's log
+       * start offset.


I read through the code and comments, it is not correct. We're saying leaderEndOffset >= replicaEndOffset case here, not leaderStratOffset. The leaderStartOffset is another case under leaderEndOffset >= replicaEndOffset`. So, maybe change to:

In the first case, [if] the follower's current log end offset is smaller than the leader's log start offset, the follower should truncate all its logs, roll out a new segment and start to fetch from the current leader's log start offset since the data are all stale.

WDYT?

I think this case refers to line 674. This comment reverts back to the original one, before the change introduced for TS. Agreed it could be made clearer though, perhaps by referencing explicitly the case below.

Yes this was just a revert to the original comments and logic. but this makes sense, I can try to clear up the cases. thanks for the comment suggestion

mattwong949 · 2023-02-14T19:59:58Z

sorry for the delay on any open comments, I'll be able to reply to and iterate on them this week

…factor

…factor and fix build on scala 2.12

junrao

@mattwong949 : Thanks for the updated PR. A few more comments.

junrao · 2023-02-23T22:13:19Z


+    def buildRemoteLog(topicPartition: TopicPartition, leaderLogStartOffset: Long): Unit = {
+      fetcher.truncateFullyAndStartAt(topicPartition, leaderState.localLogStartOffset)
+      replicaState.logStartOffset = leaderLogStartOffset


Well, buildRemoteLog() does two things (1) call fetcher.truncateFullyAndStartAt and (2) set replicaState.logStartOffset. For (1), since the truncation logic is moved to TierStateMachine, it probably should be done in MockTierStateMachine.start() directly. For (2), the existing test doesn't need to set replicaState.logStartOffset. So, it seems it's unnecessary? If address both (1) and (2), then the callback is not needed.

junrao

@mattwong949 : Thanks for the updated PR. One more followup comment.

junrao · 2023-02-24T00:30:58Z

+                         fetchPartitionData: FetchResponseData.PartitionData): PartitionFetchState = {
+        replicaState.log.clear()
+        replicaState.localLogStartOffset = 5
+        replicaState.logEndOffset = 5


Instead of the customizing for this test, I am wondering if we could make this more general. For example, could we pass in fetcher to MockTierStateMachine and call fetcher.truncateFullyAndStartAt() in MockTierStateMachine.start().

hm yeah I see what you are saying, I can make this change. the only somewhat tricky part is that we pass in the MockTierStateMachine to the fetcher itself, so we'd have to pass the reference to the fetcher to the MockTierStateMachine after both are instantiated

Got it. So, if we do that, it's the same as the callback logic you had earlier.

Here is another way to improve this. Since AbstractFetcherThread already exposes the partitionState through fetchState(), we could just get rid of replicaPartitionStates in MockFetcherThread and purely rely on the partitionState that's changed on every doWork() call. Will that work? Since that's a bigger change, we could probably just take the callback code you had in this PR and make the bigger change in a separate PR.

junrao

@mattwong949 : Thanks for the explanation. A couple of more comments.

junrao · 2023-02-24T01:08:32Z

+                         fetchPartitionData: FetchResponseData.PartitionData): PartitionFetchState = {
+        replicaState.log.clear()
+        replicaState.localLogStartOffset = 5
+        replicaState.logEndOffset = 5


Got it. So, if we do that, it's the same as the callback logic you had earlier.

Here is another way to improve this. Since AbstractFetcherThread already exposes the partitionState through fetchState(), we could just get rid of replicaPartitionStates in MockFetcherThread and purely rely on the partitionState that's changed on every doWork() call. Will that work? Since that's a bigger change, we could probably just take the callback code you had in this PR and make the bigger change in a separate PR.

junrao · 2023-02-24T01:19:48Z

+ * limitations under the License.
+ */
+
+package kafka.server;


Should this be in the storage module under org.apache.kafka.server.log.remote.storage package? Eventually, we could move ReplicaFetcherTierStateMachine to the storage module too.

yup makes sense I'll move the interface over

actually I remember why I didn't initially put it in the storage package. it seems like it wouldn't be a simple change to move this to there because the TSM relies on the PartitionFetchState case class in AbstractFetcherThread.scala. it would've created a circular dependency across modules that I wanted to avoid. @junrao wdyt about keeping it in the core module in this PR

…factor fix conflicts and change logic in ReplicaFetcherTierStateMachine

junrao

@mattwong949 : Thanks for the updated PR. LGTM

mattwong949 added 3 commits January 25, 2023 16:22

refactor logic of buildRemoteAuxLog into TierStateMachine interface. …

4894131

…also implemented in java. need to modify testing still

build fixes, checkstyle fixes, testing AbstracterFetcher related test…

63e2839

…s wip

fixed AbstractFetcherThreadTest

7851d09

mattwong949 changed the title ~~Refactor logic to handle OFFSET_MOVED_TO_TIERED_STORAGE error~~ [KAFKA-14685] Refactor logic to handle OFFSET_MOVED_TO_TIERED_STORAGE error Feb 6, 2023

rittikaadhikari reviewed Feb 7, 2023

View reviewed changes

satishd reviewed Feb 7, 2023

View reviewed changes

mattwong949 added 3 commits February 7, 2023 20:10

Merge branch 'trunk' of github.com:apache/kafka into tsm-interface-re…

42f0937

…factor

address comments

c29d421

fix imports and nits in tests

95755c8

Hangleton reviewed Feb 8, 2023

View reviewed changes

junrao reviewed Feb 8, 2023

View reviewed changes

address more comments

1742b8f

showuon reviewed Feb 10, 2023

View reviewed changes

mattwong949 added 6 commits February 17, 2023 15:46

Merge branch 'trunk' of github.com:apache/kafka into tsm-interface-re…

f937391

…factor

fixing some nits, trying to resolve comments

9610005

Merge branch 'trunk' of github.com:apache/kafka into tsm-interface-re…

1f8d636

…factor

fixing PartitionData usage from the FetchResponse

3adb0b0

Merge branch 'trunk' of github.com:apache/kafka into tsm-interface-re…

d2a3e76

…factor and fix build on scala 2.12

fix build on scala 2.12

2e38027

rittikaadhikari reviewed Feb 23, 2023

View reviewed changes

Comment thread core/src/main/scala/kafka/server/AbstractFetcherThread.scala Outdated

remove unused import

9aa8af0

junrao reviewed Feb 23, 2023

View reviewed changes

mattwong949 added 2 commits February 23, 2023 14:43

nits, will do refactor of test per comments

6525189

refactor test per Jun's comment

4b81be1

junrao reviewed Feb 24, 2023

View reviewed changes

mattwong949 added 2 commits February 23, 2023 17:26

refactor MockTierStateMachine and MockFetcherThread

f6f4416

Merge branch 'trunk' of github.com:apache/kafka into tsm-interface-re…

32d6e01

…factor fix conflicts and change logic in ReplicaFetcherTierStateMachine

fix imports and offsetAndEpoch changes from merge

5d118a0

junrao approved these changes Feb 24, 2023

View reviewed changes

junrao merged commit 8d32a0f into apache:trunk Feb 24, 2023

mdedetrich mentioned this pull request May 11, 2023

Backport Refactor logic to handle OFFSET_MOVED_TO_TIERED_STORAGE error aiven/kafka#22

Closed


		Optional<RemoteLogSegmentMetadata> maybeRlsm = rlm.fetchRemoteLogSegmentMetadata(topicPartition, targetEpoch, previousOffsetToLeaderLocalLogStartOffset);

		if (maybeRlsm.isPresent()) {

Uh oh!

Conversation

mattwong949 commented Feb 6, 2023

Uh oh!

junrao commented Feb 6, 2023

Uh oh!

rittikaadhikari left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

satishd left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Hangleton Feb 8, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mattwong949 Feb 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mattwong949 commented Feb 14, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

rittikaadhikari left a comment •

edited

Loading

satishd left a comment •

edited

Loading

Hangleton Feb 8, 2023 •

edited

Loading

mattwong949 Feb 21, 2023 •

edited

Loading

mattwong949 commented Feb 14, 2023 •

edited

Loading