You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "abhijeetk88 (via GitHub)" <gi...@apache.org> on 2024/08/14 21:36:22 UTC
[PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
abhijeetk88 opened a new pull request, #16884:
URL: https://github.com/apache/kafka/pull/16884
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16884:
URL: https://github.com/apache/kafka/pull/16884#issuecomment-3199081621
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16884:
URL: https://github.com/apache/kafka/pull/16884#discussion_r2298170907
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,551 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetching behavior when the leader
+ * log starts at 0 and no segments have been uploaded or deleted locally.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **0** (No local segments deleted).
+ *
+ * Scenario:
+ * - The leader log contains three record batches at offsets 0, 150, and 199.
+ * - The follower starts fetching from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. The replica log accurately reflects the leader's log:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 0
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled with some segments uploaded and deleted locally, causing
+ * a difference between the leader's LogStartOffset (0) and LocalLogStartOffset (> 0).
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **100** (Some segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 0 but the local leader log starts at offset 100.
+ * - The follower fetch operation begins from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. After offset adjustments for local deletions:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetch behavior when the leader's log
+ * starts at a non-zero offset and no local deletions have occurred.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **10** (No deletions).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. After fetching, the replica log matches the leader:
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and Local Deletions
+ *
+ * Purpose:
+ * - Validate follower adjustments when the leader has log deletions causing
+ * LocalLogStartOffset > LogStartOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **100** (All older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 but the local log starts at offset 100.
+ * - The follower fetch starts at offset 10 but adjusts for local deletions.
+ *
+ * Expected Outcomes:
+ * 1. Initial fetch offset adjustments:
+ * - First adjustment: LogEndOffset = 10 (after offset-out-of-range error)
+ * - Second adjustment: LogEndOffset = 100 (after offset-moved-to-tiered-storage error)
+ * 2. After successful fetches:
+ * - 3 record batches fetched
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ leaderState.logStartOffset = 10
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
Review Comment:
Done. Also added fetch-offset assertions for other tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16884:
URL: https://github.com/apache/kafka/pull/16884#issuecomment-3216191259
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16884:
URL: https://github.com/apache/kafka/pull/16884#issuecomment-2572217142
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on PR #16884:
URL: https://github.com/apache/kafka/pull/16884#issuecomment-3223155158
> Could we use ParameterizedTest to reduce the code duplicates?
We discussed offline that converting to ParameterizedTest is affecting the readability of the tests. So, we decided to keep as-is
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #16884:
URL: https://github.com/apache/kafka/pull/16884#discussion_r2299738443
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,562 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // Follower gets out-of-range error (no messages received), fetch offset is updated from 0 to 10
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetching behavior when the leader
+ * log starts at 0 and no segments have been uploaded or deleted locally.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **0** (No local segments deleted).
+ *
+ * Scenario:
+ * - The leader log contains three record batches at offsets 0, 150, and 199.
+ * - The follower starts fetching from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. The replica log accurately reflects the leader's log:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 0
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled with some segments uploaded and deleted locally, causing
+ * a difference between the leader's LogStartOffset (0) and LocalLogStartOffset (> 0).
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **100** (Some segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 0 but the local leader log starts at offset 100.
+ * - The follower fetch operation begins from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. After offset adjustments for local deletions:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetch behavior when the leader's log
+ * starts at a non-zero offset and no local deletions have occurred.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **10** (No deletions).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. After fetching, the replica log matches the leader:
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and Local Deletions
+ *
+ * Purpose:
+ * - Validate follower adjustments when the leader has log deletions causing
+ * LocalLogStartOffset > LogStartOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **100** (All older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 but the local log starts at offset 100.
+ * - The follower fetch starts at offset 10 but adjusts for local deletions.
+ *
+ * Expected Outcomes:
+ * 1. Initial fetch offset adjustments:
+ * - First adjustment: LogEndOffset = 10 (after offset-out-of-range error)
+ * - Second adjustment: LogEndOffset = 100 (after offset-moved-to-tiered-storage error)
+ * 2. After successful fetches:
+ * - 3 record batches fetched
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ leaderState.logStartOffset = 10
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ fetcher.doWork()
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, All Local Segments Deleted
+ *
+ * Purpose:
+ * - Handle scenarios where all local segments have been deleted:
+ * - LocalLogStartOffset > LogStartOffset.
+ * - LocalLogStartOffset = LogEndOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0 or > 0**
+ * - Leader LocalLogStartOffset: Leader LogEndOffset (all segments deleted locally).
+ *
+ * Expected Outcomes:
+ * 1. Follower state is adjusted to reflect local deletions:
+ * - LocalLogStartOffset = LogEndOffset.
+ * - No new data remains to fetch.
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroAllLocalSegmentsDeleted(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 151L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ // Set Local Log Start Offset to Log End Offset
+ leaderState.localLogStartOffset = 151
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ assertEquals(Some(151), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Call once again to see if new data is received
+ fetcher.doWork()
+ // No metadata update expected
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
Review Comment:
shall we also assert logStartOffset?
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,562 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // Follower gets out-of-range error (no messages received), fetch offset is updated from 0 to 10
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetching behavior when the leader
+ * log starts at 0 and no segments have been uploaded or deleted locally.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **0** (No local segments deleted).
+ *
+ * Scenario:
+ * - The leader log contains three record batches at offsets 0, 150, and 199.
+ * - The follower starts fetching from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. The replica log accurately reflects the leader's log:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 0
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled with some segments uploaded and deleted locally, causing
+ * a difference between the leader's LogStartOffset (0) and LocalLogStartOffset (> 0).
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **100** (Some segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 0 but the local leader log starts at offset 100.
+ * - The follower fetch operation begins from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. After offset adjustments for local deletions:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetch behavior when the leader's log
+ * starts at a non-zero offset and no local deletions have occurred.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **10** (No deletions).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. After fetching, the replica log matches the leader:
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and Local Deletions
+ *
+ * Purpose:
+ * - Validate follower adjustments when the leader has log deletions causing
+ * LocalLogStartOffset > LogStartOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **100** (All older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 but the local log starts at offset 100.
+ * - The follower fetch starts at offset 10 but adjusts for local deletions.
+ *
+ * Expected Outcomes:
+ * 1. Initial fetch offset adjustments:
+ * - First adjustment: LogEndOffset = 10 (after offset-out-of-range error)
+ * - Second adjustment: LogEndOffset = 100 (after offset-moved-to-tiered-storage error)
+ * 2. After successful fetches:
+ * - 3 record batches fetched
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ leaderState.logStartOffset = 10
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ fetcher.doWork()
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, All Local Segments Deleted
+ *
+ * Purpose:
+ * - Handle scenarios where all local segments have been deleted:
+ * - LocalLogStartOffset > LogStartOffset.
+ * - LocalLogStartOffset = LogEndOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0 or > 0**
+ * - Leader LocalLogStartOffset: Leader LogEndOffset (all segments deleted locally).
+ *
+ * Expected Outcomes:
+ * 1. Follower state is adjusted to reflect local deletions:
+ * - LocalLogStartOffset = LogEndOffset.
+ * - No new data remains to fetch.
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroAllLocalSegmentsDeleted(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 151L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ // Set Local Log Start Offset to Log End Offset
+ leaderState.localLogStartOffset = 151
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ assertEquals(Some(151), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Call once again to see if new data is received
+ fetcher.doWork()
+ // No metadata update expected
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and All Local Segments Deleted
+ *
+ * Purpose:
+ * - Validate follower behavior when TieredStorage is enabled, the leader's log starts at a non-zero offset,
+ * and all local log segments have been deleted.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **151** (all older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log contains record batches from offset 100, but all local segments up to offset 151 are deleted.
+ * - The follower starts at LogStartOffset = 10 and adjusts for local segment deletions.
+ *
+ * Expected Outcomes:
+ * 1. Follower detects offset adjustments due to local deletions:
+ * - LogStartOffset remains 10.
+ * - LocalLogStartOffset updates to 151.
+ * - LogEndOffset updates to 151.
+ * 2. HighWatermark aligns with the leader (151).
+ * 3. No new data is fetched since all relevant segments are deleted.
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroAllLocalSegmentsDeleted(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 151L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 10
+ // Set Local Log Start Offset to Log End Offset
+ leaderState.localLogStartOffset = 151
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ assertEquals(Some(151), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Call once again to see if new data is received
+ fetcher.doWork()
+ // No metadata update expected
+ assertEquals(0, replicaState.log.size)
Review Comment:
shall we also assert logStartOffset?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Add empty replica FollowerFetch tests [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph merged PR #16884:
URL: https://github.com/apache/kafka/pull/16884
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on PR #16884:
URL: https://github.com/apache/kafka/pull/16884#issuecomment-2585725869
I would like to revive this PR. @kamalcph @showuon @satishd will you be able to help with reviews. I can rebase the PR with trunk.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16884:
URL: https://github.com/apache/kafka/pull/16884#discussion_r2298169429
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,551 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
Review Comment:
Added a comment to clarify that the follower ran into out of range error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16884:
URL: https://github.com/apache/kafka/pull/16884#discussion_r2298167782
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,551 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
Review Comment:
Added a comment to clarify that the follower ran into Out of range error and no messages were received
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #16884:
URL: https://github.com/apache/kafka/pull/16884#discussion_r2299871505
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,562 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // Follower gets out-of-range error (no messages received), fetch offset is updated from 0 to 10
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetching behavior when the leader
+ * log starts at 0 and no segments have been uploaded or deleted locally.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **0** (No local segments deleted).
+ *
+ * Scenario:
+ * - The leader log contains three record batches at offsets 0, 150, and 199.
+ * - The follower starts fetching from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. The replica log accurately reflects the leader's log:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 0
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled with some segments uploaded and deleted locally, causing
+ * a difference between the leader's LogStartOffset (0) and LocalLogStartOffset (> 0).
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **100** (Some segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 0 but the local leader log starts at offset 100.
+ * - The follower fetch operation begins from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. After offset adjustments for local deletions:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetch behavior when the leader's log
+ * starts at a non-zero offset and no local deletions have occurred.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **10** (No deletions).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. After fetching, the replica log matches the leader:
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and Local Deletions
+ *
+ * Purpose:
+ * - Validate follower adjustments when the leader has log deletions causing
+ * LocalLogStartOffset > LogStartOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **100** (All older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 but the local log starts at offset 100.
+ * - The follower fetch starts at offset 10 but adjusts for local deletions.
+ *
+ * Expected Outcomes:
+ * 1. Initial fetch offset adjustments:
+ * - First adjustment: LogEndOffset = 10 (after offset-out-of-range error)
+ * - Second adjustment: LogEndOffset = 100 (after offset-moved-to-tiered-storage error)
+ * 2. After successful fetches:
+ * - 3 record batches fetched
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ leaderState.logStartOffset = 10
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ fetcher.doWork()
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, All Local Segments Deleted
+ *
+ * Purpose:
+ * - Handle scenarios where all local segments have been deleted:
+ * - LocalLogStartOffset > LogStartOffset.
+ * - LocalLogStartOffset = LogEndOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0 or > 0**
+ * - Leader LocalLogStartOffset: Leader LogEndOffset (all segments deleted locally).
+ *
+ * Expected Outcomes:
+ * 1. Follower state is adjusted to reflect local deletions:
+ * - LocalLogStartOffset = LogEndOffset.
+ * - No new data remains to fetch.
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroAllLocalSegmentsDeleted(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 151L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ // Set Local Log Start Offset to Log End Offset
+ leaderState.localLogStartOffset = 151
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ assertEquals(Some(151), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Call once again to see if new data is received
+ fetcher.doWork()
+ // No metadata update expected
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
Review Comment:
Added
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,562 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // Follower gets out-of-range error (no messages received), fetch offset is updated from 0 to 10
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetching behavior when the leader
+ * log starts at 0 and no segments have been uploaded or deleted locally.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **0** (No local segments deleted).
+ *
+ * Scenario:
+ * - The leader log contains three record batches at offsets 0, 150, and 199.
+ * - The follower starts fetching from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. The replica log accurately reflects the leader's log:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 0
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ assertEquals(Some(1), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled with some segments uploaded and deleted locally, causing
+ * a difference between the leader's LogStartOffset (0) and LocalLogStartOffset (> 0).
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **100** (Some segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 0 but the local leader log starts at offset 100.
+ * - The follower fetch operation begins from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. After offset adjustments for local deletions:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetch behavior when the leader's log
+ * starts at a non-zero offset and no local deletions have occurred.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **10** (No deletions).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. After fetching, the replica log matches the leader:
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and Local Deletions
+ *
+ * Purpose:
+ * - Validate follower adjustments when the leader has log deletions causing
+ * LocalLogStartOffset > LogStartOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **100** (All older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 but the local log starts at offset 100.
+ * - The follower fetch starts at offset 10 but adjusts for local deletions.
+ *
+ * Expected Outcomes:
+ * 1. Initial fetch offset adjustments:
+ * - First adjustment: LogEndOffset = 10 (after offset-out-of-range error)
+ * - Second adjustment: LogEndOffset = 100 (after offset-moved-to-tiered-storage error)
+ * 2. After successful fetches:
+ * - 3 record batches fetched
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ leaderState.logStartOffset = 10
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ fetcher.doWork()
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+ assertEquals(Some(100), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, All Local Segments Deleted
+ *
+ * Purpose:
+ * - Handle scenarios where all local segments have been deleted:
+ * - LocalLogStartOffset > LogStartOffset.
+ * - LocalLogStartOffset = LogEndOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0 or > 0**
+ * - Leader LocalLogStartOffset: Leader LogEndOffset (all segments deleted locally).
+ *
+ * Expected Outcomes:
+ * 1. Follower state is adjusted to reflect local deletions:
+ * - LocalLogStartOffset = LogEndOffset.
+ * - No new data remains to fetch.
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroAllLocalSegmentsDeleted(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 151L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ // Set Local Log Start Offset to Log End Offset
+ leaderState.localLogStartOffset = 151
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ assertEquals(Some(151), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Call once again to see if new data is received
+ fetcher.doWork()
+ // No metadata update expected
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and All Local Segments Deleted
+ *
+ * Purpose:
+ * - Validate follower behavior when TieredStorage is enabled, the leader's log starts at a non-zero offset,
+ * and all local log segments have been deleted.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **151** (all older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log contains record batches from offset 100, but all local segments up to offset 151 are deleted.
+ * - The follower starts at LogStartOffset = 10 and adjusts for local segment deletions.
+ *
+ * Expected Outcomes:
+ * 1. Follower detects offset adjustments due to local deletions:
+ * - LogStartOffset remains 10.
+ * - LocalLogStartOffset updates to 151.
+ * - LogEndOffset updates to 151.
+ * 2. HighWatermark aligns with the leader (151).
+ * 3. No new data is fetched since all relevant segments are deleted.
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroAllLocalSegmentsDeleted(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 151L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 10
+ // Set Local Log Start Offset to Log End Offset
+ leaderState.localLogStartOffset = 151
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+ assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // On offset-moved-to-tiered-storage error, fetch offset is updated
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(151, replicaState.localLogStartOffset)
+ assertEquals(151, replicaState.logEndOffset)
+ assertEquals(151, replicaState.highWatermark)
+ assertEquals(Some(151), fetcher.fetchState(partition).map(_.fetchOffset()))
+
+ // Call once again to see if new data is received
+ fetcher.doWork()
+ // No metadata update expected
+ assertEquals(0, replicaState.log.size)
Review Comment:
added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16884:
URL: https://github.com/apache/kafka/pull/16884#issuecomment-2799589199
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16884:
URL: https://github.com/apache/kafka/pull/16884#issuecomment-3208865367
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #16884:
URL: https://github.com/apache/kafka/pull/16884#discussion_r2295705981
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,551 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetching behavior when the leader
+ * log starts at 0 and no segments have been uploaded or deleted locally.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **0** (No local segments deleted).
+ *
+ * Scenario:
+ * - The leader log contains three record batches at offsets 0, 150, and 199.
+ * - The follower starts fetching from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. The replica log accurately reflects the leader's log:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 0
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(0, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset = 0, and Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled with some segments uploaded and deleted locally, causing
+ * a difference between the leader's LogStartOffset (0) and LocalLogStartOffset (> 0).
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **0**
+ * - Leader LocalLogStartOffset: **100** (Some segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 0 but the local leader log starts at offset 100.
+ * - The follower fetch operation begins from offset 0.
+ *
+ * Expected Outcomes:
+ * 1. After offset adjustments for local deletions:
+ * - LogStartOffset = 0
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ leaderState.logStartOffset = 0
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(100, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(100, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and No Local Deletions
+ *
+ * Purpose:
+ * - Simulate TieredStorage enabled and validate follower fetch behavior when the leader's log
+ * starts at a non-zero offset and no local deletions have occurred.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **10** (No deletions).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. After fetching, the replica log matches the leader:
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroNoLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 3) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.localLogStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Enabled, Leader LogStartOffset != 0, and Local Deletions
+ *
+ * Purpose:
+ * - Validate follower adjustments when the leader has log deletions causing
+ * LocalLogStartOffset > LogStartOffset.
+ *
+ * Conditions:
+ * - TieredStorage: **Enabled**
+ * - Leader LogStartOffset: **10**
+ * - Leader LocalLogStartOffset: **100** (All older segments deleted locally).
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 but the local log starts at offset 100.
+ * - The follower fetch starts at offset 10 but adjusts for local deletions.
+ *
+ * Expected Outcomes:
+ * 1. Initial fetch offset adjustments:
+ * - First adjustment: LogEndOffset = 10 (after offset-out-of-range error)
+ * - Second adjustment: LogEndOffset = 100 (after offset-moved-to-tiered-storage error)
+ * 2. After successful fetches:
+ * - 3 record batches fetched
+ * - LogStartOffset = 10
+ * - LocalLogStartOffset = 100
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageEnabledLeaderLogStartOffsetNonZeroWithLocalDeletions(): Unit = {
+ val rlmEnabled = true
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LocalLogStartOffset = 100
+ mkBatch(baseOffset = 100, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled,
+ )
+ leaderState.logStartOffset = 10
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ // On offset-out-of-range error, fetch offset is updated
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
Review Comment:
can the fetchOffset update also be asserted?
```
assertEquals(Some(10), fetcher.fetchState(partition).map(_.fetchOffset()))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17344: Empty Follower Fetch Tests [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #16884:
URL: https://github.com/apache/kafka/pull/16884#discussion_r2295697173
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,551 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
Review Comment:
could you add a comment to explain why the replica logSize is zero on first-follower-fetch when the fetch-offset is non-zero?
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1180,4 +1180,551 @@ class AbstractFetcherThreadTest {
fetcher.processFetchRequest(partitionData, fetchRequestOpt)
assertEquals(0, replicaState.logEndOffset, "FetchResponse should be ignored when leader epoch does not match")
}
+
+ private def emptyReplicaState(rlmEnabled: Boolean, partition: TopicPartition, fetcher: MockFetcherThread) = {
+ // Follower begins with an empty log
+ val replicaState = PartitionState(Seq(), leaderEpoch = 0, highWatermark = 0L, rlmEnabled = rlmEnabled)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition -> initialFetchState(topicIds.get(partition.topic), fetchOffset = 0, leaderEpoch = 0)))
+ replicaState
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset = 0
+ *
+ * Purpose:
+ * - Simulate a leader with logs starting at offset 0 and validate how the follower
+ * behaves when TieredStorage is disabled.
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **0**
+ *
+ * Scenario:
+ * - The leader starts with a log at offset 0, containing three record batches offset at 0, 150, and 199.
+ * - The follower begins fetching, and we validate the correctness of its replica state as it fetches.
+ *
+ * Expected Outcomes:
+ * 1. The follower fetch state should transition to `FETCHING` initially.
+ * 2. After the first poll, one record batch is fetched.
+ * 3. After subsequent polls, the entire leader log is fetched:
+ * - Replica log size: 3
+ * - Replica LogStartOffset: 0
+ * - Replica LogEndOffset: 200
+ * - Replica HighWatermark: 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 0
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(1, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(1, replicaState.logEndOffset)
+
+ // Only 1 record batch is returned after a poll so calling 'n' number of times to get the desired result.
+ for (_ <- 1 to 2) fetcher.doWork()
+ assertEquals(3, replicaState.log.size)
+ assertEquals(0, replicaState.logStartOffset)
+ assertEquals(200, replicaState.logEndOffset)
+ assertEquals(199, replicaState.highWatermark)
+ }
+
+ /**
+ * Test: Empty Follower Fetch with TieredStorage Disabled and Leader LogStartOffset != 0
+ *
+ * Purpose:
+ * - Validate follower behavior when the leader's log starts at a non-zero offset (10).
+ *
+ * Conditions:
+ * - TieredStorage: **Disabled**
+ * - Leader LogStartOffset: **10**
+ *
+ * Scenario:
+ * - The leader log starts at offset 10 with batches at 10, 150, and 199.
+ * - The follower starts fetching from offset 10.
+ *
+ * Expected Outcomes:
+ * 1. The follower's initial log is empty.
+ * 2. Replica offsets after polls:
+ * - LogStartOffset = 10
+ * - LogEndOffset = 200
+ * - HighWatermark = 199
+ */
+ @Test
+ def testEmptyFollowerFetchTieredStorageDisabledLeaderLogStartOffsetNonZero(): Unit = {
+ val rlmEnabled = false
+ val partition = new TopicPartition("topic1", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+
+ val replicaState = emptyReplicaState(rlmEnabled, partition, fetcher)
+
+ val leaderLog = Seq(
+ // LogStartOffset = LocalLogStartOffset = 10
+ mkBatch(baseOffset = 10, leaderEpoch = 0, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 150, leaderEpoch = 0, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 199, leaderEpoch = 0, new SimpleRecord("e".getBytes))
+ )
+
+ val leaderState = PartitionState(
+ leaderLog,
+ leaderEpoch = 0,
+ highWatermark = 199L,
+ rlmEnabled = rlmEnabled
+ )
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ assertEquals(Option(ReplicaState.FETCHING), fetcher.fetchState(partition).map(_.state))
+ assertEquals(0, replicaState.log.size)
+ assertEquals(10, replicaState.logStartOffset)
+ assertEquals(10, replicaState.logEndOffset)
Review Comment:
ditto: please add a comment for reader to understand why logStartOffset == logEndOffset on first-follower-fetch when the fetch-offset is non-zero.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org