jolshan commented on code in PR #13230:
URL: https://github.com/apache/kafka/pull/13230#discussion_r1103127795
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1089,6 +1089,85 @@ class AbstractFetcherThreadTest {
fetcher.verifyLastFetchedEpoch(partition, Some(5))
}
+ @Test
+ def testTruncateOnFetchDoesNotUpdateHighWatermark(): Unit = {
+ assumeTrue(truncateOnFetch)
+
+ val partition = new TopicPartition("topic", 0)
+
+ var truncateCalls = 0
+ var processPartitionDataCalls = 0
+ val fetcher = new MockFetcherThread(new MockLeaderEndPoint) {
+ override def processPartitionData(topicPartition: TopicPartition,
fetchOffset: Long, partitionData: FetchData): Option[LogAppendInfo] = {
+ processPartitionDataCalls += 1
+ super.processPartitionData(topicPartition, fetchOffset, partitionData)
+ }
+
+ override def truncate(topicPartition: TopicPartition, truncationState:
OffsetTruncationState): Unit = {
+ truncateCalls += 1
+ super.truncate(topicPartition, truncationState)
+ }
+ }
+
+ val replicaLog = Seq(
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("a".getBytes)),
+ mkBatch(baseOffset = 1, leaderEpoch = 0, new SimpleRecord("b".getBytes)),
+ mkBatch(baseOffset = 2, leaderEpoch = 2, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 3, leaderEpoch = 4, new SimpleRecord("d".getBytes)),
+ mkBatch(baseOffset = 4, leaderEpoch = 4, new SimpleRecord("e".getBytes)),
+ mkBatch(baseOffset = 5, leaderEpoch = 4, new SimpleRecord("f".getBytes)),
+ )
+
+ val replicaState = PartitionState(replicaLog, leaderEpoch = 5,
highWatermark = 1L)
+ fetcher.setReplicaState(partition, replicaState)
+ fetcher.addPartitions(Map(partition ->
initialFetchState(topicIds.get(partition.topic), 3L, leaderEpoch = 5)))
+ assertEquals(6L, replicaState.logEndOffset)
+ fetcher.verifyLastFetchedEpoch(partition, expectedEpoch = Some(4))
+
+ val leaderLog = Seq(
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("a".getBytes)),
+ mkBatch(baseOffset = 1, leaderEpoch = 0, new SimpleRecord("b".getBytes)),
+ mkBatch(baseOffset = 2, leaderEpoch = 2, new SimpleRecord("c".getBytes)),
+ mkBatch(baseOffset = 3, leaderEpoch = 5, new SimpleRecord("g".getBytes)),
+ mkBatch(baseOffset = 4, leaderEpoch = 5, new SimpleRecord("h".getBytes)),
+ )
+
+ val leaderState = PartitionState(leaderLog, leaderEpoch = 5, highWatermark
= 4L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+
fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ fetcher.doWork()
+ fetcher.verifyLastFetchedEpoch(partition, Some(2))
+
+ // The log end offset is truncated.
+ assertEquals(3L, replicaState.logEndOffset)
+
+ // The high watermark is not updated.
+ assertEquals(1L, replicaState.highWatermark)
+
+ // The first fetch should result in truncating the follower's log and
Review Comment:
The "first fetch" mentioned here is the fetcher.doWork() on line 1139. Would
it make sense to but the comment there? Since it applies also to the checks on
1143 and 1146.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
