junrao commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4186361729
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -345,6 +346,64 @@ class ReplicaManagerTest {
}
}
+ @Test
+ def testMaybeAddLogDirFetchersWhenFutureLogIsAheadOfHighWatermark(): Unit = {
+ val dir1 = TestUtils.tempDir()
+ val dir2 = TestUtils.tempDir()
+ val props = TestUtils.createBrokerConfig(0)
+ props.put("log.dirs", dir1.getAbsolutePath + "," + dir2.getAbsolutePath)
+ val config = KafkaConfig.fromProps(props)
+ val logManager = TestUtils.createLogManager(config.logDirs.asScala.map(new
File(_)), new LogConfig(new Properties()))
+ mockGetAliveBrokerFunctions(metadataCache, Seq(new Node(0, "host0", 0)))
+
when(metadataCache.metadataVersion()).thenReturn(MetadataVersion.MINIMUM_VERSION)
+ val rm = new ReplicaManager(
+ metrics = metrics,
+ config = config,
+ time = time,
+ scheduler = new MockScheduler(time),
+ logManager = logManager,
+ quotaManagers = quotaManager,
+ metadataCache = metadataCache,
+ logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
+ alterPartitionManager = alterPartitionManager)
+
+ try {
+ val delta = topicsCreateDelta(0, isStartIdLeader = true, partitions =
List(0), topicName = topic, topicId = topicIds(topic))
+ val image = imageFromTopics(delta.apply())
+ rm.applyDelta(delta, image)
+ val partition = rm.getPartitionOrException(topicPartition)
+ val leaderEpoch = partition.getLeaderEpoch
+
+ // Append one record per batch, so that the future log can be at a batch
boundary below the log
+ // end offset of the current replica. Fetches are batch aligned, so a
future log in the middle
+ // of a batch is not a state the fetcher can be in.
+ appendRecords(rm, topicPartition,
MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("first
message".getBytes())))
+ appendRecords(rm, topicPartition,
MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("second
message".getBytes())))
+ logManager.maybeUpdatePreferredLogDir(topicPartition,
dir2.getAbsolutePath)
+
+ partition.createLogIfNotExists(isNew = true, isFutureReplica = true,
+ new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), None)
+
+ // Simulate a future log which has already replicated a record that is
not committed on the
+ // current replica yet, so that its log end offset is ahead of its high
watermark.
+ partition.appendRecordsToFollowerOrFutureReplica(
+ MemoryRecords.withRecords(0L, Compression.NONE, leaderEpoch, new
SimpleRecord("first message".getBytes())),
+ isFuture = true, leaderEpoch)
+ val futureLog = partition.futureLocalLogOrException
+ assertTrue(futureLog.logEndOffset > futureLog.highWatermark)
+ assertTrue(futureLog.latestEpoch.isPresent)
+
+ // The fetcher has to start from the log end offset of the future log.
Starting from its high
+ // watermark makes the first fetch fail with an offset mismatch in
+ // ReplicaAlterLogDirsThread.processPartitionData, which marks the
partition as failed.
+ rm.maybeAddLogDirFetchers(Set(partition), new
LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), _ => None)
+ rm.replicaAlterLogDirsManager.fetcherThreadMap.values.foreach(t =>
t.doWork())
Review Comment:
Should we verify the initial fetch offset for the future replica before
calling doWork()?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
