hachikuji commented on a change in pull request #10935:
URL: https://github.com/apache/kafka/pull/10935#discussion_r665713457
##########
File path: core/src/test/scala/integration/kafka/server/RaftClusterTest.scala
##########
@@ -313,4 +318,94 @@ class RaftClusterTest {
cluster.close()
}
}
+
+ @Test
+ def testCreateClusterWithAdvertisedPortZero(): Unit = {
+ val nodes = new TestKitNodes.Builder()
+ .setNumControllerNodes(1)
+ .setNumBrokerNodes(3)
+ .build()
+ nodes.brokerNodes().values().forEach { broker =>
+ broker.propertyOverrides().put(KafkaConfig.ListenersProp,
+ s"${nodes.externalListenerName().value()}://localhost:0")
+ broker.propertyOverrides().put(KafkaConfig.AdvertisedListenersProp,
+ s"${nodes.externalListenerName().value()}://localhost:0")
+ }
+ val cluster = new KafkaClusterTestKit.Builder(nodes).build()
+ try {
+ cluster.format()
+ cluster.startup()
+
+ val (runningBrokerServer, foundRunningBroker) =
TestUtils.computeUntilTrue(anyBrokerServer(cluster)){
+ brokerServer => brokerServer.currentState() == BrokerState.RUNNING
+ }
+ assertTrue(foundRunningBroker, "No Broker never made it to RUNNING
state.")
Review comment:
nit: never -> ever?
##########
File path: core/src/test/scala/integration/kafka/server/RaftClusterTest.scala
##########
@@ -313,4 +318,94 @@ class RaftClusterTest {
cluster.close()
}
}
+
+ @Test
+ def testCreateClusterWithAdvertisedPortZero(): Unit = {
+ val nodes = new TestKitNodes.Builder()
+ .setNumControllerNodes(1)
+ .setNumBrokerNodes(3)
+ .build()
+ nodes.brokerNodes().values().forEach { broker =>
+ broker.propertyOverrides().put(KafkaConfig.ListenersProp,
+ s"${nodes.externalListenerName().value()}://localhost:0")
+ broker.propertyOverrides().put(KafkaConfig.AdvertisedListenersProp,
+ s"${nodes.externalListenerName().value()}://localhost:0")
+ }
+ val cluster = new KafkaClusterTestKit.Builder(nodes).build()
+ try {
+ cluster.format()
+ cluster.startup()
+
+ val (runningBrokerServer, foundRunningBroker) =
TestUtils.computeUntilTrue(anyBrokerServer(cluster)){
+ brokerServer => brokerServer.currentState() == BrokerState.RUNNING
+ }
+ assertTrue(foundRunningBroker, "No Broker never made it to RUNNING
state.")
+
+ val (describeClusterResponse, metadataUpToDate) =
TestUtils.computeUntilTrue(
+ sendDescribeClusterRequest(runningBrokerServer.socketServer,
nodes.externalListenerName())
+ ) {
+ response => response.nodes().size() == nodes.brokerNodes().size()
+ }
+ assertTrue(metadataUpToDate, s"Broker never reached expected cluster
size of ${nodes.brokerNodes().size()}")
Review comment:
nit: these tests are basically the same except for the initialization
and assertions. It would be nice to factor out a helper or two.
##########
File path: core/src/test/scala/integration/kafka/server/RaftClusterTest.scala
##########
@@ -313,4 +318,94 @@ class RaftClusterTest {
cluster.close()
}
}
+
+ @Test
+ def testCreateClusterWithAdvertisedPortZero(): Unit = {
+ val nodes = new TestKitNodes.Builder()
+ .setNumControllerNodes(1)
+ .setNumBrokerNodes(3)
+ .build()
+ nodes.brokerNodes().values().forEach { broker =>
+ broker.propertyOverrides().put(KafkaConfig.ListenersProp,
+ s"${nodes.externalListenerName().value()}://localhost:0")
+ broker.propertyOverrides().put(KafkaConfig.AdvertisedListenersProp,
+ s"${nodes.externalListenerName().value()}://localhost:0")
+ }
+ val cluster = new KafkaClusterTestKit.Builder(nodes).build()
+ try {
+ cluster.format()
+ cluster.startup()
+
+ val (runningBrokerServer, foundRunningBroker) =
TestUtils.computeUntilTrue(anyBrokerServer(cluster)){
+ brokerServer => brokerServer.currentState() == BrokerState.RUNNING
+ }
+ assertTrue(foundRunningBroker, "No Broker never made it to RUNNING
state.")
+
+ val (describeClusterResponse, metadataUpToDate) =
TestUtils.computeUntilTrue(
+ sendDescribeClusterRequest(runningBrokerServer.socketServer,
nodes.externalListenerName())
+ ) {
+ response => response.nodes().size() == nodes.brokerNodes().size()
+ }
+ assertTrue(metadataUpToDate, s"Broker never reached expected cluster
size of ${nodes.brokerNodes().size()}")
+
+ describeClusterResponse.nodes().values().forEach { node =>
+ assertEquals("localhost", node.host,
+ "Did not advertise configured advertised host")
+
assertEquals(cluster.brokers().get(node.id).socketServer.boundPort(nodes.externalListenerName()),
node.port,
+ "Did not advertise bound socket port")
+ }
+ } finally {
+ cluster.close()
+ }
+ }
+
+ @Test
+ def testCreateClusterWithAdvertisedHostAndPortDifferentFromSocketServer():
Unit = {
+ val nodes = new TestKitNodes.Builder()
+ .setNumControllerNodes(1)
+ .setNumBrokerNodes(3)
+ .build()
+ nodes.brokerNodes().values().forEach { broker =>
+ broker.propertyOverrides().put(KafkaConfig.ListenersProp,
+ s"${nodes.externalListenerName().value()}://localhost:0")
+ broker.propertyOverrides().put(KafkaConfig.AdvertisedListenersProp,
+
s"${nodes.externalListenerName().value()}://advertised-host-${broker.id}:${broker.id
+ 100}")
+ }
+ val cluster = new KafkaClusterTestKit.Builder(nodes).build()
+ try {
+ cluster.format()
+ cluster.startup()
+
+ val (runningBrokerServer, foundRunningBroker) =
TestUtils.computeUntilTrue(anyBrokerServer(cluster)){
+ brokerServer => brokerServer.currentState() == BrokerState.RUNNING
+ }
+ assertTrue(foundRunningBroker, "No Broker never made it to RUNNING
state.")
+
+ val (describeClusterResponse, metadataUpToDate) =
TestUtils.computeUntilTrue(
+ sendDescribeClusterRequest(runningBrokerServer.socketServer,
nodes.externalListenerName())
Review comment:
Maybe worth a comment that `connectAndReceive` connects to the bound
port instead of the advertised port since that is not obvious.
--
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
