[GitHub] [kafka] jsancio commented on a change in pull request #11179: KAFKA-13165: Validate node id, process role and quorum voters
jsancio commented on a change in pull request #11179: URL: https://github.com/apache/kafka/pull/11179#discussion_r685410401 ## File path: core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala ## @@ -28,27 +28,41 @@ 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.mockito.Mockito._ +import java.io.File + class RaftManagerTest { - private def instantiateRaftManagerWithConfigs(processRoles: String, nodeId:String) = { -def configWithProcessRolesAndNodeId(processRoles: String, nodeId: String): KafkaConfig = { + 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.ListenersProp, "PLAINTEXT://localhost:9093") props.setProperty(KafkaConfig.ControllerListenerNamesProp, "PLAINTEXT") - props.setProperty(RaftConfig.QUORUM_VOTERS_CONFIG, nodeId.concat("@localhost:9093")) - if (processRoles.contains("broker")) + if (processRoles.contains("broker")) { props.setProperty(KafkaConfig.InterBrokerListenerNameProp, "PLAINTEXT") props.setProperty(KafkaConfig.AdvertisedListenersProp, "PLAINTEXT://localhost:9092") +if (!processRoles.contains("controller")) { + val voterId = (nodeId.toInt + 1) + props.setProperty(KafkaConfig.QuorumVotersProp, s"${voterId}@localhost:9093") +} + } + + if (processRoles.contains("controller")) { +props.setProperty(KafkaConfig.QuorumVotersProp, s"${nodeId}@localhost:9093") + } + new KafkaConfig(props) } -val config = configWithProcessRolesAndNodeId(processRoles, nodeId) +val logDir = TestUtils.tempDirectory() Review comment: Should we remove these directories after every test? I think this doesn't delete the directory until all of the tests in `core` have finished and the JVM terminates. -- 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
[GitHub] [kafka] jsancio commented on a change in pull request #11179: KAFKA-13165: Validate node id, process role and quorum voters
jsancio commented on a change in pull request #11179: URL: https://github.com/apache/kafka/pull/11179#discussion_r685275346 ## File path: core/src/main/scala/kafka/raft/RaftManager.scala ## @@ -79,6 +80,11 @@ object KafkaRaftManager { Files.createDirectories(dir.toPath) dir } + + private def deleteLogDirectory(logDir: File, logDirName: String): Unit = { +val dir = new Directory(new File(logDir.getAbsolutePath, logDirName)) +dir.deleteRecursively() + } Review comment: This code should not be in `core/src/main` since it is not used by `core/src/main`. ## File path: core/src/main/scala/kafka/raft/RaftManager.scala ## @@ -214,6 +220,12 @@ class KafkaRaftManager[T]( KafkaRaftManager.createLogDirectory(new File(config.metadataLogDir), logDirName) } + // visible for testing cleanup + private[raft] def deleteDataDir(): Unit = { +val logDirName = Log.logDirName(topicPartition) +KafkaRaftManager.deleteLogDirectory(new File(config.metadataLogDir), logDirName) Review comment: This code should not be in `core/src/main` since it is not used by `core/src/main`. -- 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
[GitHub] [kafka] jsancio commented on a change in pull request #11179: KAFKA-13165: Validate node id, process role and quorum voters
jsancio commented on a change in pull request #11179: URL: https://github.com/apache/kafka/pull/11179#discussion_r683859800 ## File path: core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala ## @@ -262,6 +262,8 @@ class KafkaConfigTest { props.put(KafkaConfig.ProcessRolesProp, "controller") props.put(KafkaConfig.ListenersProp, "PLAINTEXT://127.0.0.1:9092") props.put(KafkaConfig.NodeIdProp, "1") +props.setProperty(RaftConfig.QUORUM_VOTERS_CONFIG, s"1@localhost:9092") Review comment: This is regular string. String interpolation `s"..."` is not needed. This comment applies to a few places. ## File path: core/src/main/scala/kafka/server/KafkaConfig.scala ## @@ -1949,6 +1951,19 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO ) } +val voterIds: Set[Integer] = if (usesSelfManagedQuorum) RaftConfig.parseVoterConnections(quorumVoters).asScala.keySet.toSet else Set.empty +if (voterIds.nonEmpty) { + if (processRoles.contains(ControllerRole)) { +// Ensure that controllers use their node.id as a voter in controller.quorum.voters +require(voterIds.contains(nodeId), s"If ${KafkaConfig.ProcessRolesProp} contains the 'controller' role, the node id $nodeId must be included in the set of voters ${RaftConfig.QUORUM_VOTERS_CONFIG}=$voterIds") + } else { +// Ensure that the broker's node.id is not an id in controller.quorum.voters +require(!voterIds.contains(nodeId), s"If ${KafkaConfig.ProcessRolesProp} does not contain the 'controller' role, the node id $nodeId must not be included in the set of voters ${RaftConfig.QUORUM_VOTERS_CONFIG}=$voterIds") + } +} else if (usesSelfManagedQuorum) { + throw new ConfigException(s"If using ${KafkaConfig.ProcessRolesProp}, ${RaftConfig.QUORUM_VOTERS_CONFIG} must contain a parseable set of voters.") +} Review comment: ```scala if (usesSelfManagedQuorum) { val voterIds = RaftConfig.parseVoterConnections(quorumVoters).asScala.keySet if (voterIds.isEmpty) { throw new ConfigException(s"If using ${KafkaConfig.ProcessRolesProp}, ${RaftConfig.QUORUM_VOTERS_CONFIG} must contain a parseable set of voters.") } else if (processRoles.contains(ControllerRole)) { // Ensure that controllers use their node.id as a voter in controller.quorum.voters require(voterIds.contains(nodeId), s"If ${KafkaConfig.ProcessRolesProp} contains the 'controller' role, the node id $nodeId must be included in the set of voters ${RaftConfig.QUORUM_VOTERS_CONFIG}=$voterIds") } else { // Ensure that the broker's node.id is not an id in controller.quorum.voters require(!voterIds.contains(nodeId), s"If ${KafkaConfig.ProcessRolesProp} does not contain the 'controller' role, the node id $nodeId must not be included in the set of voters ${RaftConfig.QUORUM_VOTERS_CONFIG}=$voterIds") } } ``` ## File path: core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala ## @@ -34,21 +34,30 @@ import org.mockito.Mockito._ class RaftManagerTest { - private def instantiateRaftManagerWithConfigs(processRoles: String, nodeId:String) = { + private def instantiateRaftManagerWithConfigs(topicPartition: TopicPartition, processRoles: String, nodeId: String) = { def configWithProcessRolesAndNodeId(processRoles: String, nodeId: String): KafkaConfig = { val props = new Properties props.setProperty(KafkaConfig.ProcessRolesProp, processRoles) props.setProperty(KafkaConfig.NodeIdProp, nodeId) props.setProperty(KafkaConfig.ListenersProp, "PLAINTEXT://localhost:9093") props.setProperty(KafkaConfig.ControllerListenerNamesProp, "PLAINTEXT") - props.setProperty(RaftConfig.QUORUM_VOTERS_CONFIG, nodeId.concat("@localhost:9093")) - if (processRoles.contains("broker")) + if (processRoles.contains("broker")) { props.setProperty(KafkaConfig.InterBrokerListenerNameProp, "PLAINTEXT") props.setProperty(KafkaConfig.AdvertisedListenersProp, "PLAINTEXT://localhost:9092") +if (!processRoles.contains("controller")) { + val nodeIdMod = (nodeId.toInt + 1) + props.setProperty(RaftConfig.QUORUM_VOTERS_CONFIG, s"${nodeIdMod.toString}@localhost:9093") Review comment: String interpolation should call `toString` so you don't need to call it. ## File path: core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala ## @@ -34,21 +34,30 @@ import org.mockito.Mockito._ class RaftManagerTest { - private def instantiateRaftManagerWithConfigs(processRoles: String, nodeId:String) = { + private def instantiateRaftManagerWithConfigs(topicPartition: TopicPartition, processRoles: String, nodeId: String) = { Review comment: The issue is that after every test you need to delete the directory
[GitHub] [kafka] jsancio commented on a change in pull request #11179: KAFKA-13165: Validate node id, process role and quorum voters
jsancio commented on a change in pull request #11179: URL: https://github.com/apache/kafka/pull/11179#discussion_r683110083 ## File path: core/src/main/scala/kafka/server/KafkaConfig.scala ## @@ -1949,6 +1951,15 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO ) } +val voterIds: Set[Integer] = if (usesSelfManagedQuorum) RaftConfig.parseVoterConnections(quorumVoters).asScala.keySet.toSet else Set.empty +if (processRoles.contains(ControllerRole)) { + // Ensure that controllers use their node.id as a voter in controller.quorum.voters + require(voterIds.contains(nodeId), s"The controller must contain a voter for it's ${KafkaConfig.NodeIdProp}=$nodeId in ${RaftConfig.QUORUM_VOTERS_CONFIG}.") +} else { + // Ensure that the broker's node.id is not an id in controller.quorum.voters + require(!voterIds.contains(nodeId), s"Since ${KafkaConfig.ProcessRolesProp}=broker, the the broker's ${KafkaConfig.NodeIdProp}=nodeId should not be in ${RaftConfig.QUORUM_VOTERS_CONFIG}") Review comment: How about "If ${process role config...} does not contain the 'controller' role, the node id $nodeId must not be included in the set of voters ${QUORUM_VOTERS_CONFIG}=$votersIds"? ## File path: core/src/main/scala/kafka/server/KafkaConfig.scala ## @@ -1949,6 +1951,15 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO ) } +val voterIds: Set[Integer] = if (usesSelfManagedQuorum) RaftConfig.parseVoterConnections(quorumVoters).asScala.keySet.toSet else Set.empty +if (processRoles.contains(ControllerRole)) { + // Ensure that controllers use their node.id as a voter in controller.quorum.voters + require(voterIds.contains(nodeId), s"The controller must contain a voter for it's ${KafkaConfig.NodeIdProp}=$nodeId in ${RaftConfig.QUORUM_VOTERS_CONFIG}.") Review comment: How about "If ${process role config...} contains the 'controller' role, the node id $nodeId must be included in the set of voters ${QUORUM_VOTERS_CONFIG}=$votersIds"? -- 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
[GitHub] [kafka] jsancio commented on a change in pull request #11179: KAFKA-13165: Validate node id, process role and quorum voters
jsancio commented on a change in pull request #11179: URL: https://github.com/apache/kafka/pull/11179#discussion_r683110083 ## File path: core/src/main/scala/kafka/server/KafkaConfig.scala ## @@ -1949,6 +1951,15 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO ) } +val voterIds: Set[Integer] = if (usesSelfManagedQuorum) RaftConfig.parseVoterConnections(quorumVoters).asScala.keySet.toSet else Set.empty +if (processRoles.contains(ControllerRole)) { + // Ensure that controllers use their node.id as a voter in controller.quorum.voters + require(voterIds.contains(nodeId), s"The controller must contain a voter for it's ${KafkaConfig.NodeIdProp}=$nodeId in ${RaftConfig.QUORUM_VOTERS_CONFIG}.") +} else { + // Ensure that the broker's node.id is not an id in controller.quorum.voters + require(!voterIds.contains(nodeId), s"Since ${KafkaConfig.ProcessRolesProp}=broker, the the broker's ${KafkaConfig.NodeIdProp}=nodeId should not be in ${RaftConfig.QUORUM_VOTERS_CONFIG}") Review comment: How about "If ${process role config...} does not contain the 'controller' role, the node id $nodeId must not be included in the set of voters ${QUORUM_VOTERS_CONFIG}=$votersIds"? ## File path: core/src/main/scala/kafka/server/KafkaConfig.scala ## @@ -1949,6 +1951,15 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO ) } +val voterIds: Set[Integer] = if (usesSelfManagedQuorum) RaftConfig.parseVoterConnections(quorumVoters).asScala.keySet.toSet else Set.empty +if (processRoles.contains(ControllerRole)) { + // Ensure that controllers use their node.id as a voter in controller.quorum.voters + require(voterIds.contains(nodeId), s"The controller must contain a voter for it's ${KafkaConfig.NodeIdProp}=$nodeId in ${RaftConfig.QUORUM_VOTERS_CONFIG}.") Review comment: How about "If ${process role config...} contains the 'controller' role, the node id $nodeId must be included in the set of voters ${QUORUM_VOTERS_CONFIG}=$votersIds"? -- 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