KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state #19223

frankvicky · 2025-03-18T06:53:25Z

This PR fixes a potential issue where the FetchResponse returns divergingEndOffsets with an older leader epoch. This can lead to committed records being removed from the follower's log, potentially causing data loss.

In detail:
processFetchRequest gets the requested leader epoch of partition data by topicPartition and compares it with the leader epoch of the current fetch state. If they don't match, the response is ignored.

… fetch request and fetch state JIRA: KAFKA-18991

junrao

@frankvicky : Thanks for the PR. The code LGTM. Could we add a test case?

frankvicky · 2025-03-20T16:05:14Z

Hi @junrao
Thanks for the review.
This is a tricky one to test.
I'm still researching how to create a condition that matches the patch's purpose.

chia7712 · 2025-03-20T18:17:03Z

core/src/main/scala/kafka/server/AbstractFetcherThread.scala

            val fetchPartitionData = sessionPartitions.get(topicPartition)
            if (fetchPartitionData != null &&
                fetchPartitionData.fetchOffset == currentFetchState.fetchOffset &&
+                fetchPartitionData.currentLeaderEpoch.map[Boolean](_ == currentFetchState.currentLeaderEpoch).orElse(true) &&


the line#368 can be streamlined to currentFetchState.currentLeaderEpoch due to this new condition.

val logAppendInfoOpt = processPartitionData( topicPartition, currentFetchState.fetchOffset, fetchPartitionData.currentLeaderEpoch.orElse(currentFetchState.currentLeaderEpoch), // this line partitionData )

junrao

@frankvicky : Thanks for the updated PR. A couple of more comments.

junrao · 2025-03-21T16:51:28Z

core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala

+    fetcher.mockLeader.setLeaderState(partition, leaderState)
+    fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+    val partitionData = Map(partition -> new FetchRequest.PartitionData(Uuid.randomUuid(), 0, 0, 1048576, Optional.of(0), Optional.of(0))).asJava


Could we use initEpoch instead of 0?

junrao · 2025-03-21T16:51:59Z

core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala

+    val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+    val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+    fetcher.mockLeader.setLeaderState(partition, leaderState)
+    fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)


These three lines seem unneeded?

I think only L1173 is redundant.

Technically, these lines don't affect the test result if initEpoch != newEpoch.
But since these lines implement the mock mechanism of fetch, these mocks will return a non-empty FetchResponse as essential safeguards:

If initEpoch is accidentally set equal to newEpoch, the response will be handled, and the leader epoch will be bumped as the leader epoch of fetchResponsethe test to fail.

If AbstractFetcherThread#L334 is accidentally removed, the response will be handled, and the leader epoch will be bumped as the leader epoch of fetchResponsethe test to fail.

Given that, we should keep these lines. WDYT?

junrao

@frankvicky : Thanks for the updated PR. Just one minor comment.

junrao · 2025-03-24T23:51:34Z

core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala

+    val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+    fetcher.mockLeader.setLeaderState(partition, leaderState)
+
+    val partitionData = Map(partition -> new FetchRequest.PartitionData(Uuid.randomUuid(), 0, 0, 1048576, Optional.of(initEpoch), Optional.of(0))).asJava


Could we use initEpoch for lastFetchedEpoch too?

junrao

@frankvicky : Thanks for the updated PR. LGTM

junrao · 2025-03-25T16:16:56Z

@frankvicky : Could you provide a PR for the 3.9 and 4.0 branch as well?

chia7712 · 2025-03-25T17:21:47Z

Could you provide a PR for the 3.9 and 4.0 branch as well?

should we backport KAFKA-18723 first?

junrao · 2025-03-25T17:45:39Z

should we backport KAFKA-18723 first?

The fixed version for KAFKA-18723 already includes 3.9.1 and 4.0.1?

chia7712 · 2025-03-25T17:52:11Z

The fixed version for KAFKA-18723 already includes 3.9.1 and 4.0.1?

I grep the git history , and KAFKA-18723 is not in 4.0/3.9.

ubuntu@cncf-ubuntu-0:~/project/kafka$ git switch 4.0
Switched to branch '4.0'
Your branch is up to date with 'origin/4.0'.
ubuntu@cncf-ubuntu-0:~/project/kafka$ git log --grep='18723'
ubuntu@cncf-ubuntu-0:~/project/kafka$ 

ubuntu@cncf-ubuntu-0:~/project/kafka$ git switch 3.9
Switched to branch '3.9'
Your branch is up to date with 'origin/3.9'.
ubuntu@cncf-ubuntu-0:~/project/kafka$ git log --grep='18723'
ubuntu@cncf-ubuntu-0:~/project/kafka$

chia7712 · 2025-03-25T18:06:57Z

I try to cherry-pick 4a8a063 to 4.0, but there are some conflicts. @jsancio do you have free cycle to file PR to cherry-pick it to 4.0?

frankvicky · 2025-03-26T03:11:07Z

I have just tried to cherry-pick this patch, but as @chia7712 mentioned.
It seems that we should pick #18852 first.

…equest and fetch state (apache#19223) This PR fixes a potential issue where the `FetchResponse` returns `divergingEndOffsets` with an older leader epoch. This can lead to committed records being removed from the follower's log, potentially causing data loss. In detail: `processFetchRequest` gets the requested leader epoch of partition data by `topicPartition` and compares it with the leader epoch of the current fetch state. If they don't match, the response is ignored. Reviewers: Jun Rao <junrao@gmail.com>

…equest and fetch state (#19223) This PR fixes a potential issue where the `FetchResponse` returns `divergingEndOffsets` with an older leader epoch. This can lead to committed records being removed from the follower's log, potentially causing data loss. In detail: `processFetchRequest` gets the requested leader epoch of partition data by `topicPartition` and compares it with the leader epoch of the current fetch state. If they don't match, the response is ignored. Reviewers: Jun Rao <junrao@gmail.com>

KAFKA-18991: AbstractFetcherThread should match leader epochs between…

671b39d

… fetch request and fetch state JIRA: KAFKA-18991

github-actions bot added triage PRs from the community core Kafka Broker small Small PRs labels Mar 18, 2025

chia7712 added the ci-approved label Mar 18, 2025

frankvicky marked this pull request as draft March 18, 2025 11:40

frankvicky added 2 commits March 18, 2025 21:30

reuse sessionPartitions

1084f25

Merge branch 'trunk' into KAFKA-18991

1c2d3e9

frankvicky marked this pull request as ready for review March 18, 2025 13:31

junrao reviewed Mar 19, 2025

View reviewed changes

Merge branch 'trunk' into KAFKA-18991

ed04c41

github-actions bot removed the triage PRs from the community label Mar 20, 2025

chia7712 reviewed Mar 20, 2025

View reviewed changes

frankvicky added 3 commits March 21, 2025 10:15

Merge branch 'trunk' into KAFKA-18991

e9ab077

Add a unit test

ee15df8

Merge branch 'trunk' into KAFKA-18991

329b5c7

junrao reviewed Mar 21, 2025

View reviewed changes

frankvicky added 4 commits March 22, 2025 07:51

Merge branch 'trunk' into KAFKA-18991

46c04f9

address by comments

bd283be

Merge branch 'trunk' into KAFKA-18991

88ff37b

Merge branch 'trunk' into KAFKA-18991

9788206

junrao reviewed Mar 24, 2025

View reviewed changes

frankvicky added 2 commits March 25, 2025 10:09

Merge branch 'trunk' into KAFKA-18991

43ba1c0

address by comments

93bf5f0

junrao approved these changes Mar 25, 2025

View reviewed changes

junrao merged commit 80d99ea into apache:trunk Mar 25, 2025
23 checks passed

KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state #19223

KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state #19223

Uh oh!

Conversation

frankvicky commented Mar 18, 2025

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Mar 20, 2025

Uh oh!

chia7712 Mar 20, 2025

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

junrao Mar 21, 2025

Choose a reason for hiding this comment

Uh oh!

junrao Mar 21, 2025

Choose a reason for hiding this comment

Uh oh!

frankvicky Mar 22, 2025 • 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!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

junrao Mar 24, 2025

Choose a reason for hiding this comment

Uh oh!

frankvicky Mar 25, 2025

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

junrao commented Mar 25, 2025

Uh oh!

chia7712 commented Mar 25, 2025

Uh oh!

junrao commented Mar 25, 2025

Uh oh!

chia7712 commented Mar 25, 2025

Uh oh!

chia7712 commented Mar 25, 2025

Uh oh!

frankvicky commented Mar 26, 2025

Uh oh!

Uh oh!

frankvicky Mar 22, 2025 •

edited

Loading