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/08/02 04:52:09 UTC
[GitHub] [kafka] dengziming opened a new pull request, #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
dengziming opened a new pull request, #12469:
URL: https://github.com/apache/kafka/pull/12469
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
Add `MetadataQuorumCommand` to describe quorum status, I'm trying to use arg4j style command format, currently, we only support one sub-command which is "describe".
```
# describe quorum status
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe
# specify AdminClient properties
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 --command-config config.properties describe
```
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
MetadataQuorumCommandTest and MetadataQuorumCommandErrorTest
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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] hachikuji merged pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji merged PR #12469:
URL: https://github.com/apache/kafka/pull/12469
--
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] hachikuji commented on pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji commented on PR #12469:
URL: https://github.com/apache/kafka/pull/12469#issuecomment-1221137115
@dengziming I reverted the patch to disable the test. I think we found the cause of the flakiness and fixed here: https://github.com/apache/kafka/pull/12508/commits/d1936aba7ac23d221b1d70a26148f0645081d61d.
--
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] hachikuji commented on a diff in pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12469:
URL: https://github.com/apache/kafka/pull/12469#discussion_r942910479
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.fileType
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers.newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser.addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser.addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ val admin = Admin.create(props)
+
+ if (command == "describe") {
+ handleDescribe(admin)
+ } else {
+ // currently we only support describe
Review Comment:
That is a fair point, but it seems like an `IllegalStateException` with a clear error can also serve as documentation for this case. In other words, it's just as easy to raise an exception as write the comment. And perhaps that gives us a tiny bit of extra protection if some day we break the parsing logic.
--
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 pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
dengziming commented on PR #12469:
URL: https://github.com/apache/kafka/pull/12469#issuecomment-1221354360
Thanks for your patience @hachikuji
--
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] hachikuji commented on pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji commented on PR #12469:
URL: https://github.com/apache/kafka/pull/12469#issuecomment-1218319853
@dengziming It looks like we need this patch in order to stabilize the new tests: https://github.com/apache/kafka/pull/12517. I will try to merge it today and trigger a rebuild.
--
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] hachikuji commented on a diff in pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12469:
URL: https://github.com/apache/kafka/pull/12469#discussion_r947175510
##########
clients/src/main/java/org/apache/kafka/common/utils/Utils.java:
##########
@@ -1451,4 +1452,49 @@ public static String[] enumOptions(Class<? extends Enum<?>> enumClass) {
.toArray(String[]::new);
}
+ private static void appendColumnValue(
+ StringBuilder rowBuilder,
+ String value,
+ int length
+ ) {
+ int padLength = length - value.length();
+ rowBuilder.append(value);
+ for (int i = 0; i < padLength; i++)
+ rowBuilder.append(' ');
+ }
+
+ private static void printRow(
+ List<Integer> columnLengths,
+ String[] row,
+ PrintStream out
+ ) {
+ StringBuilder rowBuilder = new StringBuilder();
+ for (int i = 0; i < row.length; i++) {
+ Integer columnLength = columnLengths.get(i);
+ String columnValue = row[i];
+ appendColumnValue(rowBuilder, columnValue, columnLength);
+ rowBuilder.append('\t');
+ }
+ out.println(rowBuilder);
+ }
+
+ public static void prettyPrintTable(
Review Comment:
An alternative might be to put this in `server-common` instead. We would have to add it as a dependency for `tools`, but it might be kind of nice to keep it out of `clients` if it's not really needed there.
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,172 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.utils.Utils.prettyPrintTable
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
Review Comment:
Not sure if you saw my comment. I do think it is a little nicer to raise an IllegalStateException. It gives us a little protection from the future in case we break something in the parsing logic.
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,172 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.utils.Utils.prettyPrintTable
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
+ }
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ } finally {
+ if (admin != null) {
+ admin.close()
+ }
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ val describeParser = subparsers
+ .addParser("describe")
+ .help("Describe the metadata quorum info")
+
+ val statusArgs = describeParser.addArgumentGroup("Status")
+ statusArgs
+ .addArgument("--status")
+ .help(
+ "A short summary of the quorum status and the other provides detailed information about the status of replication.")
+ .action(storeTrue())
+ val replicationArgs = describeParser.addArgumentGroup("Replication")
+ replicationArgs
+ .addArgument("--replication")
+ .help("Detailed information about the status of replication")
+ .action(storeTrue())
+ }
+
+ private def handleDescribeReplication(admin: Admin): Unit = {
+ val quorumInfo = admin.describeMetadataQuorum.quorumInfo.get
+ val leaderId = quorumInfo.leaderId
+ val leader = quorumInfo.voters.asScala.filter(_.replicaId == leaderId).head
+
+ def convertQuorumInfo(infos: Seq[QuorumInfo.ReplicaState], status: String): Seq[Array[String]] =
+ infos.map { voter =>
+ Array(voter.replicaId,
+ voter.logEndOffset,
+ leader.logEndOffset - voter.logEndOffset,
+ voter.lastFetchTimeMs.orElse(-1),
+ voter.lastCaughtUpTimeMs.orElse(-1),
+ status
+ ).map(_.toString)
+ }
+ prettyPrintTable(
+ Array("ReplicaId", "LogEndOffset", "Lag", "LastFetchTimeMs", "LastCaughtUpTimeMs", "Status"),
Review Comment:
How about `NodeId` instead of `ReplicaId` to go along with the `node.id` configuration?
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,172 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.utils.Utils.prettyPrintTable
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
+ }
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ } finally {
+ if (admin != null) {
+ admin.close()
+ }
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ val describeParser = subparsers
+ .addParser("describe")
+ .help("Describe the metadata quorum info")
+
+ val statusArgs = describeParser.addArgumentGroup("Status")
+ statusArgs
+ .addArgument("--status")
+ .help(
+ "A short summary of the quorum status and the other provides detailed information about the status of replication.")
+ .action(storeTrue())
+ val replicationArgs = describeParser.addArgumentGroup("Replication")
+ replicationArgs
+ .addArgument("--replication")
+ .help("Detailed information about the status of replication")
+ .action(storeTrue())
+ }
+
+ private def handleDescribeReplication(admin: Admin): Unit = {
+ val quorumInfo = admin.describeMetadataQuorum.quorumInfo.get
+ val leaderId = quorumInfo.leaderId
+ val leader = quorumInfo.voters.asScala.filter(_.replicaId == leaderId).head
+
+ def convertQuorumInfo(infos: Seq[QuorumInfo.ReplicaState], status: String): Seq[Array[String]] =
+ infos.map { voter =>
Review Comment:
nit: maybe rename `voter` since this function is also handling observers
##########
core/src/test/scala/unit/kafka/admin/MetadataQuorumCommandTest.scala:
##########
@@ -0,0 +1,189 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, ClusterTests, Type}
+import kafka.test.junit.ClusterTestExtensions
+import kafka.utils.TestUtils
+import org.apache.kafka.common.errors.UnsupportedVersionException
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue}
+import org.junit.jupiter.api.{Tag, Test}
+import org.junit.jupiter.api.extension.ExtendWith
+
+import java.util.concurrent.ExecutionException
+
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(clusterType = Type.KRAFT)
+@Tag("integration")
+class MetadataQuorumCommandTest(cluster: ClusterInstance) {
+
+ /**
+ * 1. The same number of broker controllers
+ * 2. More brokers than controllers
+ * 3. Fewer brokers than controllers
+ */
+ @ClusterTests(
+ Array(
+ new ClusterTest(clusterType = Type.CO_KRAFT, brokers = 3, controllers = 3),
+ new ClusterTest(clusterType = Type.KRAFT, brokers = 3, controllers = 3),
+ new ClusterTest(clusterType = Type.CO_KRAFT, brokers = 3, controllers = 4),
+ new ClusterTest(clusterType = Type.KRAFT, brokers = 3, controllers = 4),
+ new ClusterTest(clusterType = Type.CO_KRAFT, brokers = 4, controllers = 3),
+ new ClusterTest(clusterType = Type.KRAFT, brokers = 4, controllers = 3)
+ ))
+ def testDescribeQuorumReplicationSuccessful(): Unit = {
+ val describeOutput = TestUtils.grabConsoleOutput(
+ MetadataQuorumCommand.mainNoExit(
+ Array("--bootstrap-server", cluster.bootstrapServers(), "describe", "--replication"))
+ )
+
+ val leaderPattern = """\d+\s+\d+\s+\d+\s+[-]?\d+\s+[-]?\d+\s+Leader\s+""".r
+ val followerPattern = """\d+\s+\d+\s+\d+\s+[-]?\d+\s+[-]?\d+\s+Follower\s+""".r
+ val observerPattern = """\d+\s+\d+\s+\d+\s+[-]?\d+\s+[-]?\d+\s+Observer\s+""".r
+ val outputs = describeOutput.split("\n").tail
+ if (cluster.config().clusterType() == Type.CO_KRAFT) {
+ assertEquals(Math.max(cluster.config().numControllers(), cluster.config().numBrokers()), outputs.length)
+ } else {
+ assertEquals(cluster.config().numBrokers() + cluster.config().numControllers(), outputs.length)
+ }
+ // `matches` is not supported in scala 2.12, use `findFirstIn` instead.
+ assertTrue(leaderPattern.findFirstIn(outputs.head).nonEmpty)
+ assertEquals(1, outputs.count(leaderPattern.findFirstIn(_).nonEmpty))
+ assertEquals(cluster.config().numControllers() - 1, outputs.count(followerPattern.findFirstIn(_).nonEmpty))
+
+ if (cluster.config().clusterType() == Type.CO_KRAFT) {
+ assertEquals(Math.max(0, cluster.config().numBrokers() - cluster.config().numControllers()), outputs.count(observerPattern.findFirstIn(_).nonEmpty))
+ } else {
+ assertEquals(cluster.config().numBrokers(), outputs.count(observerPattern.findFirstIn(_).nonEmpty))
+ }
+ }
+
+ /**
+ * 1. The same number of broker controllers
+ * 2. More brokers than controllers
+ * 3. Fewer brokers than controllers
+ */
+ @ClusterTests(
+ Array(
+ new ClusterTest(clusterType = Type.CO_KRAFT, brokers = 3, controllers = 3),
+ new ClusterTest(clusterType = Type.KRAFT, brokers = 3, controllers = 3),
+ new ClusterTest(clusterType = Type.CO_KRAFT, brokers = 3, controllers = 4),
+ new ClusterTest(clusterType = Type.KRAFT, brokers = 3, controllers = 4),
+ new ClusterTest(clusterType = Type.CO_KRAFT, brokers = 4, controllers = 3),
+ new ClusterTest(clusterType = Type.KRAFT, brokers = 4, controllers = 3)
+ ))
+ def testDescribeQuorumStatusSuccessful(): Unit = {
+ val describeOutput = TestUtils.grabConsoleOutput(
+ MetadataQuorumCommand.mainNoExit(Array("--bootstrap-server", cluster.bootstrapServers(), "describe", "--status"))
+ )
+ val outputs = describeOutput.split("\n")
+
+ assertTrue("""ClusterId:\s+\S{22}""".r.findFirstIn(outputs(0)).nonEmpty)
+ assertTrue("""LeaderId:\s+\d+""".r.findFirstIn(outputs(1)).nonEmpty)
+ assertTrue("""LeaderEpoch:\s+\d+""".r.findFirstIn(outputs(2)).nonEmpty)
+ assertTrue("""HighWatermark:\s+\d+""".r.findFirstIn(outputs(3)).nonEmpty)
+ assertTrue("""MaxFollowerLag:\s+\d+""".r.findFirstIn(outputs(4)).nonEmpty)
+ assertTrue("""MaxFollowerLagTimeMs:\s+[-]?\d+""".r.findFirstIn(outputs(5)).nonEmpty)
+ assertTrue("""CurrentVoters:\s+\[\d+(,\d+)*\]""".r.findFirstIn(outputs(6)).nonEmpty)
+
+ // There are no observers if we have fewer brokers than controllers
+ if (cluster.config().clusterType() == Type.CO_KRAFT
+ && cluster.config().numBrokers() <= cluster.config().numControllers()) {
+ assertTrue("""CurrentObservers:\s+\[\]""".r.findFirstIn(outputs(7)).nonEmpty)
+ } else {
+ assertTrue("""CurrentObservers:\s+\[\d+(,\d+)*\]""".r.findFirstIn(outputs(7)).nonEmpty)
+ }
+ }
+
+ @ClusterTests(
+ Array(new ClusterTest(clusterType = Type.CO_KRAFT, brokers = 1, controllers = 1),
+ new ClusterTest(clusterType = Type.KRAFT, brokers = 1, controllers = 1)))
+ def testOnlyOneBrokerAndOneController(): Unit = {
+ val statusOutput = TestUtils.grabConsoleOutput(
+ MetadataQuorumCommand.mainNoExit(Array("--bootstrap-server", cluster.bootstrapServers(), "describe", "--status"))
+ )
+ assertEquals("MaxFollowerLag: 0", statusOutput.split("\n")(4))
+ assertEquals("MaxFollowerLagTimeMs: 0", statusOutput.split("\n")(5))
+
+ val replicationOutput = TestUtils.grabConsoleOutput(
+ MetadataQuorumCommand.mainNoExit(Array("--bootstrap-server", cluster.bootstrapServers(), "describe", "--replication"))
+ )
+ assertEquals("0", replicationOutput.split("\n").last.split("\\s+")(2))
+ }
+
+ @ClusterTest(clusterType = Type.ZK, brokers = 3, controllers = 1)
Review Comment:
nit: is it necessary to specify `controllers` when the type is ZK?
##########
core/src/test/java/kafka/test/ClusterInstance.java:
##########
@@ -115,6 +116,8 @@ default Optional<ListenerName> controlPlaneListenerName() {
*/
Map<Integer, BrokerFeatures> brokerFeatures();
+ Collection<ControllerServer> controllerServers();
Review Comment:
Perhaps it would be better not to expose this here if it is not supported by both implementations. We can still cast the `ClusterInstance` to `RaftClusterInstance` to get access to the method.
--
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 #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12469:
URL: https://github.com/apache/kafka/pull/12469#discussion_r946478935
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
+ }
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ } finally {
+ if (admin != null) {
+ admin.close()
+ }
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ val describeParser = subparsers
+ .addParser("describe")
+ .help("Describe the metadata quorum info")
+
+ val statusArgs = describeParser.addArgumentGroup("Status")
+ statusArgs
+ .addArgument("--status")
+ .help(
+ "A short summary of the quorum status and the other provides detailed information about the status of replication.")
+ .action(storeTrue())
+ val replicationArgs = describeParser.addArgumentGroup("Replication")
+ replicationArgs
+ .addArgument("--replication")
+ .help("Detailed information about the status of replication")
+ .action(storeTrue())
+ }
+
+ private def handleDescribeReplication(admin: Admin): Unit = {
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
+ val leader = quorumInfo.voters().asScala.filter(_.replicaId() == leaderId).head
+ // Find proper columns width
+ var (maxReplicaIdLen, maxLogEndOffsetLen, maxLagLen, maxLastFetchTimeMsLen, maxLastCaughtUpTimeMsLen) =
Review Comment:
`TransactionsCommand` is in a separate module, I moved this method to `Utils` in the `clients` module and adjusted the method.
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
+ }
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ } finally {
+ if (admin != null) {
+ admin.close()
+ }
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ val describeParser = subparsers
+ .addParser("describe")
+ .help("Describe the metadata quorum info")
+
+ val statusArgs = describeParser.addArgumentGroup("Status")
+ statusArgs
+ .addArgument("--status")
+ .help(
+ "A short summary of the quorum status and the other provides detailed information about the status of replication.")
+ .action(storeTrue())
+ val replicationArgs = describeParser.addArgumentGroup("Replication")
+ replicationArgs
+ .addArgument("--replication")
+ .help("Detailed information about the status of replication")
+ .action(storeTrue())
+ }
+
+ private def handleDescribeReplication(admin: Admin): Unit = {
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
+ val leader = quorumInfo.voters().asScala.filter(_.replicaId() == leaderId).head
+ // Find proper columns width
+ var (maxReplicaIdLen, maxLogEndOffsetLen, maxLagLen, maxLastFetchTimeMsLen, maxLastCaughtUpTimeMsLen) =
+ (15, 15, 15, 15, 18)
+ (quorumInfo.voters().asScala ++ quorumInfo.observers().asScala).foreach { voter =>
+ maxReplicaIdLen = Math.max(maxReplicaIdLen, voter.replicaId().toString.length)
+ maxLogEndOffsetLen = Math.max(maxLogEndOffsetLen, voter.logEndOffset().toString.length)
+ maxLagLen = Math.max(maxLagLen, (leader.logEndOffset() - voter.logEndOffset()).toString.length)
+ maxLastFetchTimeMsLen = Math.max(maxLastFetchTimeMsLen, leader.lastFetchTimeMs().orElse(-1).toString.length)
+ maxLastCaughtUpTimeMsLen =
+ Math.max(maxLastCaughtUpTimeMsLen, leader.lastCaughtUpTimeMs().orElse(-1).toString.length)
+ }
+ println(
+ s"%${-maxReplicaIdLen}s %${-maxLogEndOffsetLen}s %${-maxLagLen}s %${-maxLastFetchTimeMsLen}s %${-maxLastCaughtUpTimeMsLen}s %-15s "
+ .format("ReplicaId", "LogEndOffset", "Lag", "LastFetchTimeMs", "LastCaughtUpTimeMs", "Status")
+ )
+
+ def printQuorumInfo(infos: Seq[QuorumInfo.ReplicaState], status: String): Unit =
+ infos.foreach { voter =>
+ println(
+ s"%${-maxReplicaIdLen}s %${-maxLogEndOffsetLen}s %${-maxLagLen}s %${-maxLastFetchTimeMsLen}s %${-maxLastCaughtUpTimeMsLen}s %-15s "
+ .format(
+ voter.replicaId(),
+ voter.logEndOffset(),
+ leader.logEndOffset() - voter.logEndOffset(),
+ voter.lastFetchTimeMs().orElse(-1),
+ voter.lastCaughtUpTimeMs().orElse(-1),
+ status
+ ))
+ }
+ printQuorumInfo(Seq(leader), "Leader")
+ printQuorumInfo(quorumInfo.voters().asScala.filter(_.replicaId() != leaderId).toSeq, "Follower")
+ printQuorumInfo(quorumInfo.observers().asScala.toSeq, "Observer")
+ }
+
+ private def handleDescribeStatus(admin: Admin): Unit = {
+ val clusterId = admin.describeCluster().clusterId().get()
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
+ val leader = quorumInfo.voters().asScala.filter(_.replicaId() == leaderId).head
+ val maxLagFollower = quorumInfo
+ .voters()
+ .asScala
+ .filter(_.replicaId() != leaderId)
+ .minBy(_.logEndOffset())
Review Comment:
Good catch, I also added a method `testOnlyOneBrokerAndOneController` for this case.
--
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] hachikuji commented on a diff in pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12469:
URL: https://github.com/apache/kafka/pull/12469#discussion_r941548893
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.fileType
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers.newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser.addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser.addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ val admin = Admin.create(props)
+
+ if (command == "describe") {
+ handleDescribe(admin)
+ } else {
+ // currently we only support describe
+ }
+ admin.close()
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ subparsers.addParser("describe")
+ .help("Describe the metadata quorum info")
+ }
+
+ def handleDescribe(admin: Admin): Unit = {
Review Comment:
The output seems a bit different from what was documented in KIP-595 (and KIP-836):
- https://cwiki.apache.org/confluence/display/KAFKA/KIP-595%3A+A+Raft+Protocol+for+the+Metadata+Quorum#KIP595:ARaftProtocolfortheMetadataQuorum-ToolingSupport.
- https://cwiki.apache.org/confluence/display/KAFKA/KIP-836:+Addition+of+Information+in+DescribeQuorumResponse+about+Voter+Lag
Any reason to change it?
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.fileType
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers.newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser.addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser.addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ val admin = Admin.create(props)
+
+ if (command == "describe") {
+ handleDescribe(admin)
+ } else {
+ // currently we only support describe
Review Comment:
Should we throw an exception?
--
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] hachikuji commented on a diff in pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12469:
URL: https://github.com/apache/kafka/pull/12469#discussion_r946063842
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
+ }
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ } finally {
+ if (admin != null) {
+ admin.close()
+ }
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ val describeParser = subparsers
+ .addParser("describe")
+ .help("Describe the metadata quorum info")
+
+ val statusArgs = describeParser.addArgumentGroup("Status")
+ statusArgs
+ .addArgument("--status")
+ .help(
+ "A short summary of the quorum status and the other provides detailed information about the status of replication.")
+ .action(storeTrue())
+ val replicationArgs = describeParser.addArgumentGroup("Replication")
+ replicationArgs
+ .addArgument("--replication")
+ .help("Detailed information about the status of replication")
+ .action(storeTrue())
+ }
+
+ private def handleDescribeReplication(admin: Admin): Unit = {
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
+ val leader = quorumInfo.voters().asScala.filter(_.replicaId() == leaderId).head
+ // Find proper columns width
+ var (maxReplicaIdLen, maxLogEndOffsetLen, maxLagLen, maxLastFetchTimeMsLen, maxLastCaughtUpTimeMsLen) =
+ (15, 15, 15, 15, 18)
+ (quorumInfo.voters().asScala ++ quorumInfo.observers().asScala).foreach { voter =>
+ maxReplicaIdLen = Math.max(maxReplicaIdLen, voter.replicaId().toString.length)
+ maxLogEndOffsetLen = Math.max(maxLogEndOffsetLen, voter.logEndOffset().toString.length)
+ maxLagLen = Math.max(maxLagLen, (leader.logEndOffset() - voter.logEndOffset()).toString.length)
+ maxLastFetchTimeMsLen = Math.max(maxLastFetchTimeMsLen, leader.lastFetchTimeMs().orElse(-1).toString.length)
+ maxLastCaughtUpTimeMsLen =
+ Math.max(maxLastCaughtUpTimeMsLen, leader.lastCaughtUpTimeMs().orElse(-1).toString.length)
+ }
+ println(
+ s"%${-maxReplicaIdLen}s %${-maxLogEndOffsetLen}s %${-maxLagLen}s %${-maxLastFetchTimeMsLen}s %${-maxLastCaughtUpTimeMsLen}s %-15s "
+ .format("ReplicaId", "LogEndOffset", "Lag", "LastFetchTimeMs", "LastCaughtUpTimeMs", "Status")
+ )
+
+ def printQuorumInfo(infos: Seq[QuorumInfo.ReplicaState], status: String): Unit =
+ infos.foreach { voter =>
+ println(
+ s"%${-maxReplicaIdLen}s %${-maxLogEndOffsetLen}s %${-maxLagLen}s %${-maxLastFetchTimeMsLen}s %${-maxLastCaughtUpTimeMsLen}s %-15s "
+ .format(
+ voter.replicaId(),
+ voter.logEndOffset(),
+ leader.logEndOffset() - voter.logEndOffset(),
+ voter.lastFetchTimeMs().orElse(-1),
+ voter.lastCaughtUpTimeMs().orElse(-1),
+ status
+ ))
+ }
+ printQuorumInfo(Seq(leader), "Leader")
+ printQuorumInfo(quorumInfo.voters().asScala.filter(_.replicaId() != leaderId).toSeq, "Follower")
+ printQuorumInfo(quorumInfo.observers().asScala.toSeq, "Observer")
+ }
+
+ private def handleDescribeStatus(admin: Admin): Unit = {
+ val clusterId = admin.describeCluster().clusterId().get()
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
Review Comment:
nit: it's conventional to leave off parenthesis for getters in Scala. there are a few of these in here.
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
+ }
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ } finally {
+ if (admin != null) {
+ admin.close()
+ }
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ val describeParser = subparsers
+ .addParser("describe")
+ .help("Describe the metadata quorum info")
+
+ val statusArgs = describeParser.addArgumentGroup("Status")
+ statusArgs
+ .addArgument("--status")
+ .help(
+ "A short summary of the quorum status and the other provides detailed information about the status of replication.")
+ .action(storeTrue())
+ val replicationArgs = describeParser.addArgumentGroup("Replication")
+ replicationArgs
+ .addArgument("--replication")
+ .help("Detailed information about the status of replication")
+ .action(storeTrue())
+ }
+
+ private def handleDescribeReplication(admin: Admin): Unit = {
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
+ val leader = quorumInfo.voters().asScala.filter(_.replicaId() == leaderId).head
+ // Find proper columns width
+ var (maxReplicaIdLen, maxLogEndOffsetLen, maxLagLen, maxLastFetchTimeMsLen, maxLastCaughtUpTimeMsLen) =
+ (15, 15, 15, 15, 18)
+ (quorumInfo.voters().asScala ++ quorumInfo.observers().asScala).foreach { voter =>
+ maxReplicaIdLen = Math.max(maxReplicaIdLen, voter.replicaId().toString.length)
+ maxLogEndOffsetLen = Math.max(maxLogEndOffsetLen, voter.logEndOffset().toString.length)
+ maxLagLen = Math.max(maxLagLen, (leader.logEndOffset() - voter.logEndOffset()).toString.length)
+ maxLastFetchTimeMsLen = Math.max(maxLastFetchTimeMsLen, leader.lastFetchTimeMs().orElse(-1).toString.length)
+ maxLastCaughtUpTimeMsLen =
+ Math.max(maxLastCaughtUpTimeMsLen, leader.lastCaughtUpTimeMs().orElse(-1).toString.length)
+ }
+ println(
+ s"%${-maxReplicaIdLen}s %${-maxLogEndOffsetLen}s %${-maxLagLen}s %${-maxLastFetchTimeMsLen}s %${-maxLastCaughtUpTimeMsLen}s %-15s "
+ .format("ReplicaId", "LogEndOffset", "Lag", "LastFetchTimeMs", "LastCaughtUpTimeMs", "Status")
+ )
+
+ def printQuorumInfo(infos: Seq[QuorumInfo.ReplicaState], status: String): Unit =
+ infos.foreach { voter =>
+ println(
+ s"%${-maxReplicaIdLen}s %${-maxLogEndOffsetLen}s %${-maxLagLen}s %${-maxLastFetchTimeMsLen}s %${-maxLastCaughtUpTimeMsLen}s %-15s "
+ .format(
+ voter.replicaId(),
+ voter.logEndOffset(),
+ leader.logEndOffset() - voter.logEndOffset(),
+ voter.lastFetchTimeMs().orElse(-1),
+ voter.lastCaughtUpTimeMs().orElse(-1),
+ status
+ ))
+ }
+ printQuorumInfo(Seq(leader), "Leader")
+ printQuorumInfo(quorumInfo.voters().asScala.filter(_.replicaId() != leaderId).toSeq, "Follower")
+ printQuorumInfo(quorumInfo.observers().asScala.toSeq, "Observer")
+ }
+
+ private def handleDescribeStatus(admin: Admin): Unit = {
+ val clusterId = admin.describeCluster().clusterId().get()
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
+ val leader = quorumInfo.voters().asScala.filter(_.replicaId() == leaderId).head
+ val maxLagFollower = quorumInfo
+ .voters()
+ .asScala
+ .filter(_.replicaId() != leaderId)
+ .minBy(_.logEndOffset())
Review Comment:
This raises an error if there is only one voter. I think we can just get rid of the `filter`? Then the max follower lag is just 0 if there is only one voter. We need another adjustment for `maxFollowerLagTimeMs` so that it also ends up as 0 when there is only one voter.
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.{fileType, storeTrue}
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers
+ .newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser
+ .addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser
+ .addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ var admin: Admin = null
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ admin = Admin.create(props)
+
+ if (command == "describe") {
+ if (namespace.getBoolean("status") && namespace.getBoolean("replication")) {
+ throw new TerseFailure(s"Only one of --status or --replication should be specified with describe sub-command")
+ } else if (namespace.getBoolean("replication")) {
+ handleDescribeReplication(admin)
+ } else if (namespace.getBoolean("status")) {
+ handleDescribeStatus(admin)
+ } else {
+ throw new TerseFailure(s"One of --status or --replication must be specified with describe sub-command")
+ }
+ } else {
+ // currently we only support describe
+ }
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ } finally {
+ if (admin != null) {
+ admin.close()
+ }
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ val describeParser = subparsers
+ .addParser("describe")
+ .help("Describe the metadata quorum info")
+
+ val statusArgs = describeParser.addArgumentGroup("Status")
+ statusArgs
+ .addArgument("--status")
+ .help(
+ "A short summary of the quorum status and the other provides detailed information about the status of replication.")
+ .action(storeTrue())
+ val replicationArgs = describeParser.addArgumentGroup("Replication")
+ replicationArgs
+ .addArgument("--replication")
+ .help("Detailed information about the status of replication")
+ .action(storeTrue())
+ }
+
+ private def handleDescribeReplication(admin: Admin): Unit = {
+ val quorumInfo = admin.describeMetadataQuorum().quorumInfo().get()
+ val leaderId = quorumInfo.leaderId()
+ val leader = quorumInfo.voters().asScala.filter(_.replicaId() == leaderId).head
+ // Find proper columns width
+ var (maxReplicaIdLen, maxLogEndOffsetLen, maxLagLen, maxLastFetchTimeMsLen, maxLastCaughtUpTimeMsLen) =
Review Comment:
Potentially we could use `TransactionsCommand.prettyPrintTable` to do the formatting.
--
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 #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12469:
URL: https://github.com/apache/kafka/pull/12469#discussion_r942341166
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.fileType
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers.newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser.addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser.addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ val admin = Admin.create(props)
+
+ if (command == "describe") {
+ handleDescribe(admin)
+ } else {
+ // currently we only support describe
+ }
+ admin.close()
+ 0
+ } catch {
+ case e: TerseFailure =>
+ Console.err.println(e.getMessage)
+ 1
+ }
+ }
+
+ def addDescribeParser(subparsers: Subparsers): Unit = {
+ subparsers.addParser("describe")
+ .help("Describe the metadata quorum info")
+ }
+
+ def handleDescribe(admin: Admin): Unit = {
Review Comment:
Thank you for reminding this, I forgot to follow to format in the documentations.
##########
core/src/main/scala/kafka/admin/MetadataQuorumCommand.scala:
##########
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.admin
+
+import kafka.tools.TerseFailure
+import kafka.utils.Exit
+import net.sourceforge.argparse4j.ArgumentParsers
+import net.sourceforge.argparse4j.impl.Arguments.fileType
+import net.sourceforge.argparse4j.inf.Subparsers
+import org.apache.kafka.clients._
+import org.apache.kafka.clients.admin.{Admin, QuorumInfo}
+import org.apache.kafka.common.utils.Utils
+
+import java.io.File
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+
+/**
+ * A tool for describing quorum status
+ */
+object MetadataQuorumCommand {
+
+ def main(args: Array[String]): Unit = {
+ val res = mainNoExit(args)
+ Exit.exit(res)
+ }
+
+ def mainNoExit(args: Array[String]): Int = {
+ val parser = ArgumentParsers.newArgumentParser("kafka-metadata-quorum")
+ .defaultHelp(true)
+ .description("This tool describes kraft metadata quorum status.")
+ parser.addArgument("--bootstrap-server")
+ .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
+ .required(true)
+
+ parser.addArgument("--command-config")
+ .`type`(fileType())
+ .help("Property file containing configs to be passed to Admin Client.")
+ val subparsers = parser.addSubparsers().dest("command")
+ addDescribeParser(subparsers)
+
+ try {
+ val namespace = parser.parseArgsOrFail(args)
+ val command = namespace.getString("command")
+
+ val commandConfig = namespace.get[File]("command_config")
+ val props = if (commandConfig != null) {
+ if (!commandConfig.exists()) {
+ throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
+ }
+ Utils.loadProps(commandConfig.getPath)
+ } else {
+ new Properties()
+ }
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
+ val admin = Admin.create(props)
+
+ if (command == "describe") {
+ handleDescribe(admin)
+ } else {
+ // currently we only support describe
Review Comment:
We would fail when executing `parser.parseArgsOrFail(args)` if we specify a different command, so we won't enter this else branch at all, so it seems redundant to throw an exception 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] hachikuji commented on pull request #12469: KAFKA-13914: Add command line tool kafka-metadata-quorum.sh
Posted by GitBox <gi...@apache.org>.
hachikuji commented on PR #12469:
URL: https://github.com/apache/kafka/pull/12469#issuecomment-1220878950
@dengziming I've disabled `MetadataQuorumCommandTest` for now so that we can get this checked in. I see a similar issue in https://github.com/apache/kafka/pull/12469, but I am unable to reproduce it locally. Let's follow up to try and get to the bottom of it so that we can enable the test.
--
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