You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/05/24 20:39:35 UTC
[GitHub] [kafka] cmccabe opened a new pull request, #12205: MINOR: Fix missing UNREGISTER_BROKER call in KafkaApis
cmccabe opened a new pull request, #12205:
URL: https://github.com/apache/kafka/pull/12205
KafkaApis should forward UnregisterBroker to the KRaft 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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12205: MINOR: Fix missing UNREGISTER_BROKER call in KafkaApis
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12205:
URL: https://github.com/apache/kafka/pull/12205#discussion_r883064741
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -725,4 +726,56 @@ class KRaftClusterTest {
cluster.close()
}
}
+
+ private def clusterImage(
+ cluster: KafkaClusterTestKit,
+ brokerId: Int
+ ): ClusterImage = {
+ cluster.brokers().get(brokerId).metadataCache.currentImage().cluster()
+ }
+
+ private def brokerIsUnfenced(
+ image: ClusterImage,
+ brokerId: Int
+ ): Boolean = {
+ Option(image.brokers().get(brokerId)) match {
+ case None => false
+ case Some(registration) => !registration.fenced()
+ }
+ }
+
+ private def brokerIsAbsent(
+ image: ClusterImage,
+ brokerId: Int
+ ): Boolean = {
+ Option(image.brokers().get(brokerId)).isEmpty
+ }
+
+ @Test
+ def testUnregisterBroker(): Unit = {
+ val cluster = new KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder().
+ setNumBrokerNodes(4).
+ setNumControllerNodes(3).build()).build()
+ try {
+ cluster.format()
+ cluster.startup()
+ cluster.waitForReadyBrokers()
+ TestUtils.waitUntilTrue(() => brokerIsUnfenced(clusterImage(cluster, 1), 0),
+ "Timed out waiting for broker 0 to be unfenced.")
+ cluster.brokers().get(0).shutdown()
+ TestUtils.waitUntilTrue(() => !brokerIsUnfenced(clusterImage(cluster, 1), 0),
+ "Timed out waiting for broker 0 to be fenced.")
+ val admin = Admin.create(cluster.clientProperties())
+ try {
Review Comment:
Hmm, the try..catch block seems more like normal Scala style to me. So I'll leave it as-is for now...
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12205: MINOR: Fix missing UNREGISTER_BROKER call in KafkaApis
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12205:
URL: https://github.com/apache/kafka/pull/12205#discussion_r882331721
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -229,6 +229,8 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.ENVELOPE => handleEnvelope(request, requestLocal)
case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request)
case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request)
+ case ApiKeys.UNREGISTER_BROKER => maybeForwardToController(request,
Review Comment:
good idea
--
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
[GitHub] [kafka] dengziming commented on a diff in pull request #12205: MINOR: Fix missing UNREGISTER_BROKER call in KafkaApis
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12205:
URL: https://github.com/apache/kafka/pull/12205#discussion_r881253148
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -229,6 +229,8 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.ENVELOPE => handleEnvelope(request, requestLocal)
case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request)
case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request)
+ case ApiKeys.UNREGISTER_BROKER => maybeForwardToController(request,
Review Comment:
nit: we already have a `forwardToControllerOrFail` method which can be used here.
--
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
[GitHub] [kafka] cmccabe merged pull request #12205: MINOR: Fix missing UNREGISTER_BROKER call in KafkaApis
Posted by GitBox <gi...@apache.org>.
cmccabe merged PR #12205:
URL: https://github.com/apache/kafka/pull/12205
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #12205: MINOR: Fix missing UNREGISTER_BROKER call in KafkaApis
Posted by GitBox <gi...@apache.org>.
dajac commented on code in PR #12205:
URL: https://github.com/apache/kafka/pull/12205#discussion_r881435578
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -229,6 +229,8 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.ENVELOPE => handleEnvelope(request, requestLocal)
case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request)
case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request)
+ case ApiKeys.UNREGISTER_BROKER => maybeForwardToController(request,
Review Comment:
+1
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -725,4 +726,56 @@ class KRaftClusterTest {
cluster.close()
}
}
+
+ private def clusterImage(
+ cluster: KafkaClusterTestKit,
+ brokerId: Int
+ ): ClusterImage = {
+ cluster.brokers().get(brokerId).metadataCache.currentImage().cluster()
+ }
+
+ private def brokerIsUnfenced(
+ image: ClusterImage,
+ brokerId: Int
+ ): Boolean = {
+ Option(image.brokers().get(brokerId)) match {
+ case None => false
+ case Some(registration) => !registration.fenced()
+ }
+ }
+
+ private def brokerIsAbsent(
+ image: ClusterImage,
+ brokerId: Int
+ ): Boolean = {
+ Option(image.brokers().get(brokerId)).isEmpty
+ }
+
+ @Test
+ def testUnregisterBroker(): Unit = {
+ val cluster = new KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder().
+ setNumBrokerNodes(4).
+ setNumControllerNodes(3).build()).build()
+ try {
+ cluster.format()
+ cluster.startup()
+ cluster.waitForReadyBrokers()
+ TestUtils.waitUntilTrue(() => brokerIsUnfenced(clusterImage(cluster, 1), 0),
+ "Timed out waiting for broker 0 to be unfenced.")
+ cluster.brokers().get(0).shutdown()
+ TestUtils.waitUntilTrue(() => !brokerIsUnfenced(clusterImage(cluster, 1), 0),
+ "Timed out waiting for broker 0 to be fenced.")
+ val admin = Admin.create(cluster.clientProperties())
+ try {
Review Comment:
nit: We could use `TestUtils.resource` for auto-closing the admin and the cluster. Not a big win so I leave it up to you.
--
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