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

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

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


##########
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:
   I wanted to test having different `metadata.log.dir` and `log.dirs` with the broker. The matrix in `testLogDirLockWhenControllerOnly` already tests having different `metadata.log.dir` and `log.dirs` for the controller.



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