You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2022/01/26 07:25:36 UTC
[kafka] branch 3.1 updated: KAFKA-13585; Fix flaky test `ReplicaManagerTest.testReplicaAlterLogDirsWithAndWithoutIds` (#11665)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch 3.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.1 by this push:
new 56bc731 KAFKA-13585; Fix flaky test `ReplicaManagerTest.testReplicaAlterLogDirsWithAndWithoutIds` (#11665)
56bc731 is described below
commit 56bc731bfe1d6fc9bc5bc555a2e3a94bea4528bf
Author: David Jacot <dj...@confluent.io>
AuthorDate: Wed Jan 26 08:14:00 2022 +0100
KAFKA-13585; Fix flaky test `ReplicaManagerTest.testReplicaAlterLogDirsWithAndWithoutIds` (#11665)
The issue was quite subtile. It was due to a race for the `partitionMapLock` lock. `assertFetcherHasTopicId` would only succeed if it can acquire it before `processFetchRequest`. This PR refactors the test in order to make it more stable.
Reviewers: Justine Olshan <jo...@confluent.io>, Jason Gustafson <ja...@confluent.io>,
---
.../unit/kafka/server/ReplicaManagerTest.scala | 100 ++++++++++++---------
1 file changed, 58 insertions(+), 42 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 2b1a397..07a4351 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -2009,7 +2009,8 @@ class ReplicaManagerTest {
brokerId: Int = 0,
aliveBrokerIds: Seq[Int] = Seq(0, 1),
propsModifier: Properties => Unit = _ => {},
- mockReplicaFetcherManager: Option[ReplicaFetcherManager] = None
+ mockReplicaFetcherManager: Option[ReplicaFetcherManager] = None,
+ mockReplicaAlterLogDirsManager: Option[ReplicaAlterLogDirsManager] = None
): ReplicaManager = {
val props = TestUtils.createBrokerConfig(brokerId, TestUtils.MockZkConnect)
props.put("log.dirs", TestUtils.tempRelativeDir("data").getAbsolutePath + "," + TestUtils.tempRelativeDir("data2").getAbsolutePath)
@@ -2064,6 +2065,18 @@ class ReplicaManagerTest {
)
}
}
+
+ override def createReplicaAlterLogDirsManager(
+ quotaManager: ReplicationQuotaManager,
+ brokerTopicStats: BrokerTopicStats
+ ): ReplicaAlterLogDirsManager = {
+ mockReplicaAlterLogDirsManager.getOrElse {
+ super.createReplicaAlterLogDirsManager(
+ quotaManager,
+ brokerTopicStats
+ )
+ }
+ }
}
}
@@ -2814,7 +2827,7 @@ class ReplicaManagerTest {
// Delete the data directory to trigger a storage exception
Utils.delete(dataDir)
- val request = leaderAndIsrRequest(
+ val request = makeLeaderAndIsrRequest(
topicId = Uuid.randomUuid(),
topicPartition = topicPartition,
replicas = Seq(0, 1),
@@ -2830,7 +2843,7 @@ class ReplicaManagerTest {
}
}
- private def leaderAndIsrRequest(
+ private def makeLeaderAndIsrRequest(
topicId: Uuid,
topicPartition: TopicPartition,
replicas: Seq[Int],
@@ -2887,7 +2900,7 @@ class ReplicaManagerTest {
// This API is supported by both leaders and followers
val barPartition = new TopicPartition("bar", 0)
- val barLeaderAndIsrRequest = leaderAndIsrRequest(
+ val barLeaderAndIsrRequest = makeLeaderAndIsrRequest(
topicId = Uuid.randomUuid(),
topicPartition = barPartition,
replicas = Seq(brokerId),
@@ -2899,7 +2912,7 @@ class ReplicaManagerTest {
val otherBrokerId = 1
val bazPartition = new TopicPartition("baz", 0)
- val bazLeaderAndIsrRequest = leaderAndIsrRequest(
+ val bazLeaderAndIsrRequest = makeLeaderAndIsrRequest(
topicId = Uuid.randomUuid(),
topicPartition = bazPartition,
replicas = Seq(brokerId, otherBrokerId),
@@ -3515,7 +3528,7 @@ class ReplicaManagerTest {
// or does not start with a topic ID in the PartitionFetchState and adds one on the next request (!startsWithTopicId)
val startingId = if (startsWithTopicId) topicId else Uuid.ZERO_UUID
val startingIdOpt = if (startsWithTopicId) Some(topicId) else None
- val leaderAndIsrRequest1 = leaderAndIsrRequest(startingId, tp, aliveBrokersIds, leaderAndIsr)
+ val leaderAndIsrRequest1 = makeLeaderAndIsrRequest(startingId, tp, aliveBrokersIds, leaderAndIsr)
val leaderAndIsrResponse1 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
assertEquals(Errors.NONE, leaderAndIsrResponse1.error)
@@ -3523,7 +3536,7 @@ class ReplicaManagerTest {
val endingId = if (!startsWithTopicId) topicId else Uuid.ZERO_UUID
val endingIdOpt = if (!startsWithTopicId) Some(topicId) else None
- val leaderAndIsrRequest2 = leaderAndIsrRequest(endingId, tp, aliveBrokersIds, leaderAndIsr)
+ val leaderAndIsrRequest2 = makeLeaderAndIsrRequest(endingId, tp, aliveBrokersIds, leaderAndIsr)
val leaderAndIsrResponse2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest2, (_, _) => ())
assertEquals(Errors.NONE, leaderAndIsrResponse2.error)
@@ -3536,48 +3549,51 @@ class ReplicaManagerTest {
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testReplicaAlterLogDirsWithAndWithoutIds(usesTopicIds: Boolean): Unit = {
+ val tp = new TopicPartition(topic, 0)
val version = if (usesTopicIds) LeaderAndIsrRequestData.HIGHEST_SUPPORTED_VERSION else 4.toShort
val topicId = if (usesTopicIds) this.topicId else Uuid.ZERO_UUID
val topicIdOpt = if (usesTopicIds) Some(topicId) else None
- val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time))
- try {
- val topicPartition = new TopicPartition(topic, 0)
- val aliveBrokersIds = Seq(0, 1)
- replicaManager.createPartition(topicPartition)
- .createLogIfNotExists(isNew = false, isFutureReplica = false,
- new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints), None)
- val tp = new TopicPartition(topic, 0)
- val leaderAndIsr = new LeaderAndIsr(0, 0, aliveBrokersIds.toList, 0)
-
- val leaderAndIsrRequest1 = leaderAndIsrRequest(topicId, tp, aliveBrokersIds, leaderAndIsr, version = version)
- replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
- val partition = replicaManager.getPartitionOrException(tp)
- assertEquals(1, replicaManager.logManager.liveLogDirs.filterNot(_ == partition.log.get.dir.getParentFile).size)
-
- // Append a couple of messages.
- for (i <- 1 to 500) {
- val records = TestUtils.singletonRecords(s"message $i".getBytes)
- appendRecords(replicaManager, tp, records).onFire { response =>
- assertEquals(Errors.NONE, response.error)
- }
- }
- // Find the live and different folder.
- val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == partition.log.get.dir.getParentFile).head
- assertEquals(0, replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.size)
- replicaManager.alterReplicaLogDirs(Map(topicPartition -> newReplicaFolder.getAbsolutePath))
+ val mockReplicaAlterLogDirsManager = Mockito.mock(classOf[ReplicaAlterLogDirsManager])
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(
+ timer = new MockTimer(time),
+ mockReplicaAlterLogDirsManager = Some(mockReplicaAlterLogDirsManager)
+ )
- assertFetcherHasTopicId(replicaManager.replicaAlterLogDirsManager, partition.topicPartition, topicIdOpt)
+ try {
+ replicaManager.createPartition(tp).createLogIfNotExists(
+ isNew = false,
+ isFutureReplica = false,
+ offsetCheckpoints = new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints),
+ topicId = None
+ )
- // Make sure the future log is created.
- replicaManager.futureLocalLogOrException(topicPartition)
- assertEquals(1, replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.size)
+ val leaderAndIsrRequest = makeLeaderAndIsrRequest(
+ topicId = topicId,
+ topicPartition = tp,
+ replicas = Seq(0, 1),
+ leaderAndIsr = LeaderAndIsr(0, List(0, 1)),
+ version = version
+ )
+ replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest, (_, _) => ())
- // Wait for the ReplicaAlterLogDirsThread to complete.
- TestUtils.waitUntilTrue(() => {
- replicaManager.replicaAlterLogDirsManager.shutdownIdleFetcherThreads()
- replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.isEmpty
- }, s"ReplicaAlterLogDirsThread should be gone")
+ // Move the replica to the second log directory.
+ val partition = replicaManager.getPartitionOrException(tp)
+ val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == partition.log.get.dir.getParentFile).head
+ replicaManager.alterReplicaLogDirs(Map(tp -> newReplicaFolder.getAbsolutePath))
+
+ // Make sure the future log is created with the correct topic ID.
+ val futureLog = replicaManager.futureLocalLogOrException(tp)
+ assertEquals(topicIdOpt, futureLog.topicId)
+
+ // Verify that addFetcherForPartitions was called with the correct topic ID.
+ Mockito.verify(mockReplicaAlterLogDirsManager, Mockito.times(1))
+ .addFetcherForPartitions(Map(tp -> InitialFetchState(
+ topicId = topicIdOpt,
+ leader = BrokerEndPoint(0, "localhost", -1),
+ currentLeaderEpoch = 0,
+ initOffset = 0
+ )))
} finally {
replicaManager.shutdown(checkpointHW = false)
}