rondagostino commented on code in PR #13058:
URL: https://github.com/apache/kafka/pull/13058#discussion_r1064790638
##########
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
+ }
Review Comment:
I think this might be clearer.
```
@ValueSource(strings = Array("metadata-only", "log-only", "both"))
def testLogDirLockWhenControllerOnly(dirType: String): Unit = {
val logDir = if (!dirType.equals("metadata-only")) {
Some(TestUtils.tempDir().toPath)
} else {
None
}
val metadataDir = if (!dirType.equals("log-only")) {
Some(TestUtils.tempDir().toPath)
} else {
None
}
```
##########
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 = {
Review Comment:
A better name might be `testLogDirLockWhenBrokerOnlyWithSeparateMetadataDir`
##########
core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala:
##########
@@ -16,52 +16,71 @@
*/
package kafka.raft
-import java.util.concurrent.CompletableFuture
+import java.nio.channels.FileChannel
+import java.nio.channels.OverlappingFileLockException
+import java.nio.file.Path
+import java.nio.file.StandardOpenOption
import java.util.Properties
+import java.util.concurrent.CompletableFuture
+import kafka.log.LogManager
import kafka.raft.KafkaRaftManager.RaftIoThread
-import kafka.server.{KafkaConfig, MetaProperties}
+import kafka.server.KafkaConfig
+import kafka.server.KafkaRaftServer.BrokerRole
+import kafka.server.KafkaRaftServer.ControllerRole
+import kafka.server.KafkaRaftServer.ProcessRole
+import kafka.server.MetaProperties
+import kafka.utils.TestUtils
import kafka.tools.TestRaftServer.ByteArraySerde
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.utils.Time
import org.apache.kafka.raft.KafkaRaftClient
import org.apache.kafka.raft.RaftConfig
-import org.apache.kafka.test.TestUtils
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
import org.mockito.Mockito._
-import java.io.File
-
class RaftManagerTest {
-
- private def instantiateRaftManagerWithConfigs(topicPartition:
TopicPartition, processRoles: String, nodeId: String) = {
- def configWithProcessRolesAndNodeId(processRoles: String, nodeId: String,
logDir: File): KafkaConfig = {
- val props = new Properties
- props.setProperty(KafkaConfig.MetadataLogDirProp, logDir.getPath)
- props.setProperty(KafkaConfig.ProcessRolesProp, processRoles)
- props.setProperty(KafkaConfig.NodeIdProp, nodeId)
- props.setProperty(KafkaConfig.ControllerListenerNamesProp, "SSL")
- if (processRoles.contains("broker")) {
- props.setProperty(KafkaConfig.InterBrokerListenerNameProp, "PLAINTEXT")
- if (processRoles.contains("controller")) { // co-located
- props.setProperty(KafkaConfig.ListenersProp,
"PLAINTEXT://localhost:9092,SSL://localhost:9093")
- props.setProperty(KafkaConfig.QuorumVotersProp,
s"${nodeId}@localhost:9093")
- } else { // broker-only
- val voterId = (nodeId.toInt + 1)
- props.setProperty(KafkaConfig.QuorumVotersProp,
s"${voterId}@localhost:9093")
- }
- } else if (processRoles.contains("controller")) { // controller-only
- props.setProperty(KafkaConfig.ListenersProp, "SSL://localhost:9093")
+ private def createConfig(
+ processRoles: Set[ProcessRole],
+ nodeId: Int,
+ logDir: Option[Path],
+ metadataDir: Option[Path]
+ ): KafkaConfig = {
+ val props = new Properties
+ logDir.foreach { value =>
+ props.setProperty(KafkaConfig.LogDirProp, value.toString)
+ }
+ metadataDir.foreach { value =>
+ props.setProperty(KafkaConfig.MetadataLogDirProp, value.toString)
+ }
+ props.setProperty(KafkaConfig.ProcessRolesProp, processRoles.mkString(","))
+ props.setProperty(KafkaConfig.NodeIdProp, nodeId.toString)
+ props.setProperty(KafkaConfig.ControllerListenerNamesProp, "SSL")
+ if (processRoles.contains(BrokerRole)) {
+ props.setProperty(KafkaConfig.InterBrokerListenerNameProp, "PLAINTEXT")
+ if (processRoles.contains(ControllerRole)) { // co-located
+ props.setProperty(KafkaConfig.ListenersProp,
"PLAINTEXT://localhost:9092,SSL://localhost:9093")
props.setProperty(KafkaConfig.QuorumVotersProp,
s"${nodeId}@localhost:9093")
+ } else { // broker-only
+ val voterId = (nodeId.toInt + 1)
Review Comment:
`s/(nodeId.toInt + 1)/nodeId + 1/`
--
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
