[GitHub] [kafka] dengziming commented on a diff in pull request #13058: KAFKA-14557; Lock metadata log dir

[GitBox <gi...@apache.org>]

dengziming commented on code in PR #13058:
URL: https://github.com/apache/kafka/pull/13058#discussion_r1059224316


##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -81,25 +100,99 @@ class RaftManagerTest {
     )
   }
 
-  @Test
-  def testNodeIdPresentIfBrokerRoleOnly(): Unit = {
-    val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "broker", "1")
-    assertEquals(1, raftManager.client.nodeId.getAsInt)
+  @ParameterizedTest
+  @ValueSource(strings = Array("broker", "controller", "broker,controller"))
+  def testNodeIdPresent(processRoles: String): Unit = {
+    var processRolesSet = Set.empty[ProcessRole]
+    if (processRoles.contains("broker")) {
+      processRolesSet = processRolesSet ++ Set(BrokerRole)
+    }
+    if (processRoles.contains("controller")) {
+      processRolesSet = processRolesSet ++ Set(ControllerRole)
+    }
+
+    val logDir = TestUtils.tempDir()
+    val nodeId = 1
+    val raftManager = createRaftManager(
+      new TopicPartition("__raft_id_test", 0),
+      createConfig(
+        processRolesSet,
+        nodeId,
+        Some(logDir.toPath),
+        None
+      )
+    )
+    assertEquals(nodeId, raftManager.client.nodeId.getAsInt)
     raftManager.shutdown()
   }
 
-  @Test
-  def testNodeIdPresentIfControllerRoleOnly(): Unit = {
-    val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller", "1")
-    assertEquals(1, raftManager.client.nodeId.getAsInt)
+  @ParameterizedTest
+  @ValueSource(strings = Array("metadata", "log", "metadata,log"))
+  def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
+    val logDir = if (dirType.contains("metadata")) {
+      Some(TestUtils.tempDir().toPath)
+    } else {
+      None
+    }
+
+    val metadataDir = if (dirType.contains("log")) {
+      Some(TestUtils.tempDir().toPath)
+    } else {
+      None
+    }
+
+    val nodeId = 1
+    val raftManager = createRaftManager(
+      new TopicPartition("__raft_id_test", 0),
+      createConfig(
+        Set(ControllerRole),
+        nodeId,
+        logDir,
+        metadataDir
+      )
+    )
+
+    val lockPath = metadataDir.getOrElse(logDir.get).resolve(LogManager.LockFileName)
+    assertTrue(fileLocked(lockPath))
+
     raftManager.shutdown()
+
+    assertFalse(fileLocked(lockPath))
   }
 
   @Test
-  def testNodeIdPresentIfColocated(): Unit = {
-    val raftManager = instantiateRaftManagerWithConfigs(new TopicPartition("__raft_id_test", 0), "controller,broker", "1")
-    assertEquals(1, raftManager.client.nodeId.getAsInt)
+  def testLogDirLockWhenMetadataDir(): Unit = {
+    val logDir = Some(TestUtils.tempDir().toPath)
+    val metadataDir = Some(TestUtils.tempDir().toPath)
+
+    val nodeId = 1
+    val raftManager = createRaftManager(
+      new TopicPartition("__raft_id_test", 0),
+      createConfig(
+        Set(BrokerRole),

Review Comment:
   We are testing controller error so should this be ControllerRole.



##########
core/src/main/scala/kafka/raft/RaftManager.scala:
##########
@@ -154,6 +192,8 @@ class KafkaRaftManager[T](
     scheduler.shutdown()
     netChannel.close()
     replicatedLog.close()
+
+    dataDirLock.foreach(_.destroy())

Review Comment:
   In `LogManager` we do this in a finally block, should we do it here?



-- 
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