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
created. This is where raft creates the directory
https://github.com/apache/kafka/blob/9bc45d4e031bef63772b3d8d7c5828c0733a27b0/core/src/main/scala/kafka/raft/RaftManager.scala#L212-L215
So you need to override that with a temp dir from
https://github.com/apache/kafka/blob/9bc45d4e031bef63772b3d8d7c5828c0733a27b0/core/src/test/scala/unit/kafka/utils/TestUtils.scala#L117
and delete it after every test.
##########
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")
+ }
+ }
+
+ if (processRoles.contains("controller")) {
+ props.setProperty(RaftConfig.QUORUM_VOTERS_CONFIG,
s"${nodeId}@localhost:9093")
+ }
+
new KafkaConfig(props)
}
- val config = configWithProcessRolesAndNodeId(processRoles, nodeId)
+ val config = configWithProcessRolesAndNodeId(processRoles, nodeId)
Review comment:
Extra space at the end.
##########
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")
Review comment:
In the interest of making it consistent with all the `KafkaConfig`
properties shouldn't we have an variable for `RaftConfig.QUORUM_VOTERS_CONFIG`?
Maybe `KafkaConfig.QuorumVotersProp`. This comment applies to a few places in
this change.
##########
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)
Review comment:
Looks like this is the voter id so `val voterId = ...`
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -132,6 +133,8 @@ class KafkaApisTest {
val properties = TestUtils.createBrokerConfig(brokerId, "")
properties.put(KafkaConfig.NodeIdProp, brokerId.toString)
properties.put(KafkaConfig.ProcessRolesProp, "broker")
+ val nodeIdMod = (brokerId + 1)
+ properties.setProperty(RaftConfig.QUORUM_VOTERS_CONFIG,
s"${nodeIdMod.toString}@localhost:9093")
Review comment:
String interpolation should call `toString`. This comment applies to a
few places.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
