You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/08/22 01:22:26 UTC
(kafka) branch trunk updated: KAFKA-16324 Move BrokerApiVersionsCommand to tools (#16850)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 28cd7136fb1 KAFKA-16324 Move BrokerApiVersionsCommand to tools (#16850)
28cd7136fb1 is described below
commit 28cd7136fb1b5702af82c3736d246c7a1e709140
Author: PoAn Yang <pa...@apache.org>
AuthorDate: Thu Aug 22 09:22:20 2024 +0800
KAFKA-16324 Move BrokerApiVersionsCommand to tools (#16850)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
bin/kafka-broker-api-versions.sh | 2 +-
bin/windows/kafka-broker-api-versions.bat | 2 +-
.../kafka/admin/BrokerApiVersionsCommand.scala | 332 ---------------------
.../kafka/admin/BrokerApiVersionsCommandTest.scala | 107 -------
.../kafka/tools/BrokerApiVersionsCommand.java | 296 ++++++++++++++++++
.../kafka/tools/BrokerApiVersionsCommandTest.java | 106 +++++++
6 files changed, 404 insertions(+), 441 deletions(-)
diff --git a/bin/kafka-broker-api-versions.sh b/bin/kafka-broker-api-versions.sh
index 4f560a0a60c..285a9488af0 100755
--- a/bin/kafka-broker-api-versions.sh
+++ b/bin/kafka-broker-api-versions.sh
@@ -14,4 +14,4 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.BrokerApiVersionsCommand "$@"
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.BrokerApiVersionsCommand "$@"
diff --git a/bin/windows/kafka-broker-api-versions.bat b/bin/windows/kafka-broker-api-versions.bat
index f7ec72da55f..7c0444bc06f 100644
--- a/bin/windows/kafka-broker-api-versions.bat
+++ b/bin/windows/kafka-broker-api-versions.bat
@@ -14,4 +14,4 @@ rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
rem See the License for the specific language governing permissions and
rem limitations under the License.
-%~dp0kafka-run-class.bat kafka.admin.BrokerApiVersionsCommand %*
+%~dp0kafka-run-class.bat org.apache.kafka.tools.BrokerApiVersionsCommand %*
diff --git a/core/src/main/scala/kafka/admin/BrokerApiVersionsCommand.scala b/core/src/main/scala/kafka/admin/BrokerApiVersionsCommand.scala
deleted file mode 100644
index a8b7c8f59b5..00000000000
--- a/core/src/main/scala/kafka/admin/BrokerApiVersionsCommand.scala
+++ /dev/null
@@ -1,332 +0,0 @@
-/**
- * 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 java.io.PrintStream
-import java.io.IOException
-import java.util.Properties
-import java.util.concurrent.atomic.AtomicInteger
-import java.util.concurrent.{ConcurrentLinkedQueue, TimeUnit}
-import joptsimple.OptionSpec
-import kafka.utils.Implicits._
-import kafka.utils.Logging
-import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.clients.{ApiVersions, ClientDnsLookup, ClientResponse, ClientUtils, CommonClientConfigs, Metadata, MetadataRecoveryStrategy, NetworkClient, NodeApiVersions}
-import org.apache.kafka.clients.consumer.internals.{ConsumerNetworkClient, RequestFuture}
-import org.apache.kafka.common.config.ConfigDef.ValidString._
-import org.apache.kafka.common.config.ConfigDef.{Importance, Type}
-import org.apache.kafka.common.config.{AbstractConfig, ConfigDef}
-import org.apache.kafka.common.errors.AuthenticationException
-import org.apache.kafka.common.internals.ClusterResourceListeners
-import org.apache.kafka.common.metrics.Metrics
-import org.apache.kafka.common.network.Selector
-import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.utils.LogContext
-import org.apache.kafka.common.utils.{KafkaThread, Time}
-import org.apache.kafka.common.Node
-import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, ApiVersionsRequest, ApiVersionsResponse, MetadataRequest, MetadataResponse}
-import org.apache.kafka.common.security.auth.SecurityProtocol
-import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
-
-import scala.jdk.CollectionConverters._
-import scala.util.{Failure, Success, Try}
-
-/**
- * A command for retrieving broker version information.
- */
-object BrokerApiVersionsCommand {
-
- def main(args: Array[String]): Unit = {
- execute(args, System.out)
- }
-
- def execute(args: Array[String], out: PrintStream): Unit = {
- val opts = new BrokerVersionCommandOptions(args)
- val adminClient = createAdminClient(opts)
- adminClient.awaitBrokers()
- val brokerMap = adminClient.listAllBrokerVersionInfo()
- brokerMap.forKeyValue { (broker, versionInfoOrError) =>
- versionInfoOrError match {
- case Success(v) => out.print(s"$broker -> ${v.toString(true)}\n")
- case Failure(v) => out.print(s"$broker -> ERROR: $v\n")
- }
- }
- adminClient.close()
- }
-
- private def createAdminClient(opts: BrokerVersionCommandOptions): AdminClient = {
- val props = if (opts.options.has(opts.commandConfigOpt))
- Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt))
- else
- new Properties()
- props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt))
- AdminClient.create(props)
- }
-
- private class BrokerVersionCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) {
- private val BootstrapServerDoc = "REQUIRED: The server to connect to."
- private val CommandConfigDoc = "A property file containing configs to be passed to Admin Client."
-
- val commandConfigOpt: OptionSpec[String] = parser.accepts("command-config", CommandConfigDoc)
- .withRequiredArg
- .describedAs("command config property file")
- .ofType(classOf[String])
- val bootstrapServerOpt: OptionSpec[String] = parser.accepts("bootstrap-server", BootstrapServerDoc)
- .withRequiredArg
- .describedAs("server(s) to use for bootstrapping")
- .ofType(classOf[String])
- options = parser.parse(args : _*)
- checkArgs()
-
- private def checkArgs(): Unit = {
- CommandLineUtils.maybePrintHelpOrVersion(this, "This tool helps to retrieve broker version information.")
- // check required args
- CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt)
- }
- }
-
- // org.apache.kafka.clients.admin.AdminClient doesn't currently expose a way to retrieve the supported api versions.
- // We inline the bits we need from kafka.admin.AdminClient so that we can delete it.
- private class AdminClient(val time: Time,
- val client: ConsumerNetworkClient,
- val bootstrapBrokers: List[Node]) extends Logging {
-
- @volatile private var running = true
- private val pendingFutures = new ConcurrentLinkedQueue[RequestFuture[ClientResponse]]()
-
- private val networkThread = new KafkaThread("admin-client-network-thread", () => {
- try {
- while (running)
- client.poll(time.timer(Long.MaxValue))
- } catch {
- case t: Throwable =>
- error("admin-client-network-thread exited", t)
- } finally {
- pendingFutures.forEach { future =>
- try {
- future.raise(Errors.UNKNOWN_SERVER_ERROR)
- } catch {
- case _: IllegalStateException => // It is OK if the future has been completed
- }
- }
- pendingFutures.clear()
- }
- }, true)
-
- networkThread.start()
-
- private def send(target: Node,
- request: AbstractRequest.Builder[_ <: AbstractRequest]): AbstractResponse = {
- val future = client.send(target, request)
- pendingFutures.add(future)
- future.awaitDone(Long.MaxValue, TimeUnit.MILLISECONDS)
- pendingFutures.remove(future)
- if (future.succeeded())
- future.value().responseBody()
- else
- throw future.exception()
- }
-
- private def sendAnyNode(request: AbstractRequest.Builder[_ <: AbstractRequest]): AbstractResponse = {
- bootstrapBrokers.foreach { broker =>
- try {
- return send(broker, request)
- } catch {
- case e: AuthenticationException =>
- throw e
- case e: Exception =>
- debug(s"Request ${request.apiKey()} failed against node $broker", e)
- }
- }
- throw new RuntimeException(s"Request ${request.apiKey()} failed on brokers $bootstrapBrokers")
- }
-
- private def getNodeApiVersions(node: Node): NodeApiVersions = {
- val response = send(node, new ApiVersionsRequest.Builder()).asInstanceOf[ApiVersionsResponse]
- Errors.forCode(response.data.errorCode).maybeThrow()
- new NodeApiVersions(response.data.apiKeys, response.data.supportedFeatures, response.data.zkMigrationReady)
- }
-
- /**
- * Wait until there is a non-empty list of brokers in the cluster.
- */
- def awaitBrokers(): Unit = {
- var nodes = List[Node]()
- do {
- nodes = findAllBrokers()
- if (nodes.isEmpty)
- Thread.sleep(50)
- } while (nodes.isEmpty)
- }
-
- private def findAllBrokers(): List[Node] = {
- val request = MetadataRequest.Builder.allTopics()
- val response = sendAnyNode(request).asInstanceOf[MetadataResponse]
- val errors = response.errors
- if (!errors.isEmpty)
- debug(s"Metadata request contained errors: $errors")
- response.buildCluster.nodes.asScala.toList
- }
-
- def listAllBrokerVersionInfo(): Map[Node, Try[NodeApiVersions]] =
- findAllBrokers().map { broker =>
- broker -> Try[NodeApiVersions](getNodeApiVersions(broker))
- }.toMap
-
- def close(): Unit = {
- running = false
- try {
- client.close()
- } catch {
- case e: IOException =>
- error("Exception closing nioSelector:", e)
- }
- }
-
- }
-
- private object AdminClient {
- private val DefaultConnectionMaxIdleMs = 9 * 60 * 1000
- private val DefaultRequestTimeoutMs = 5000
- private val DefaultMaxInFlightRequestsPerConnection = 100
- private val DefaultReconnectBackoffMs = 50
- private val DefaultReconnectBackoffMax = 50
- private val DefaultSendBufferBytes = 128 * 1024
- private val DefaultReceiveBufferBytes = 32 * 1024
- private val DefaultRetryBackoffMs = 100
-
- private val AdminClientIdSequence = new AtomicInteger(1)
- private val AdminConfigDef = {
- val config = new ConfigDef()
- .define(
- CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
- Type.LIST,
- Importance.HIGH,
- CommonClientConfigs.BOOTSTRAP_SERVERS_DOC)
- .define(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG,
- Type.STRING,
- ClientDnsLookup.USE_ALL_DNS_IPS.toString,
- in(ClientDnsLookup.USE_ALL_DNS_IPS.toString,
- ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString),
- Importance.MEDIUM,
- CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
- .define(
- CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
- ConfigDef.Type.STRING,
- CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL,
- ConfigDef.CaseInsensitiveValidString.in(Utils.enumOptions(classOf[SecurityProtocol]):_*),
- ConfigDef.Importance.MEDIUM,
- CommonClientConfigs.SECURITY_PROTOCOL_DOC)
- .define(
- CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG,
- ConfigDef.Type.INT,
- DefaultRequestTimeoutMs,
- ConfigDef.Importance.MEDIUM,
- CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC)
- .define(
- CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG,
- ConfigDef.Type.LONG,
- CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MS,
- ConfigDef.Importance.MEDIUM,
- CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_DOC)
- .define(
- CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG,
- ConfigDef.Type.LONG,
- CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS,
- ConfigDef.Importance.MEDIUM,
- CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_DOC)
- .define(
- CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG,
- ConfigDef.Type.LONG,
- DefaultRetryBackoffMs,
- ConfigDef.Importance.MEDIUM,
- CommonClientConfigs.RETRY_BACKOFF_MS_DOC)
- .withClientSslSupport()
- .withClientSaslSupport()
- config
- }
-
- class AdminConfig(originals: Map[_,_]) extends AbstractConfig(AdminConfigDef, originals.asJava, Utils.castToStringObjectMap(originals.asJava), false)
-
- def create(props: Properties): AdminClient = create(props.asScala.toMap)
-
- def create(props: Map[String, _]): AdminClient = create(new AdminConfig(props))
-
- def create(config: AdminConfig): AdminClient = {
- val clientId = "admin-" + AdminClientIdSequence.getAndIncrement()
- val logContext = new LogContext(s"[LegacyAdminClient clientId=$clientId] ")
- val time = Time.SYSTEM
- val metrics = new Metrics(time)
- val metadata = new Metadata(CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MS,
- CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MAX_MS,
- 60 * 60 * 1000L, logContext,
- new ClusterResourceListeners)
- val channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext)
- val requestTimeoutMs = config.getInt(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG)
- val connectionSetupTimeoutMs = config.getLong(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG)
- val connectionSetupTimeoutMaxMs = config.getLong(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG)
- val retryBackoffMs = config.getLong(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG)
-
- val brokerUrls = config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)
- val clientDnsLookup = config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG)
- val brokerAddresses = ClientUtils.parseAndValidateAddresses(brokerUrls, clientDnsLookup)
- metadata.bootstrap(brokerAddresses)
-
- val selector = new Selector(
- DefaultConnectionMaxIdleMs,
- metrics,
- time,
- "admin",
- channelBuilder,
- logContext)
-
- val networkClient = new NetworkClient(
- selector,
- metadata,
- clientId,
- DefaultMaxInFlightRequestsPerConnection,
- DefaultReconnectBackoffMs,
- DefaultReconnectBackoffMax,
- DefaultSendBufferBytes,
- DefaultReceiveBufferBytes,
- requestTimeoutMs,
- connectionSetupTimeoutMs,
- connectionSetupTimeoutMaxMs,
- time,
- true,
- new ApiVersions,
- logContext,
- MetadataRecoveryStrategy.NONE)
-
- val highLevelClient = new ConsumerNetworkClient(
- logContext,
- networkClient,
- metadata,
- time,
- retryBackoffMs,
- requestTimeoutMs,
- Integer.MAX_VALUE)
-
- new AdminClient(
- time,
- highLevelClient,
- metadata.fetch.nodes.asScala.toList)
- }
- }
-
-}
diff --git a/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala b/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala
deleted file mode 100644
index 90f719dff8c..00000000000
--- a/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * 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.integration.KafkaServerTestHarness
-import kafka.server.KafkaConfig
-import kafka.utils.TestUtils
-import org.apache.kafka.clients.NodeApiVersions
-import org.apache.kafka.common.message.ApiMessageType
-import org.apache.kafka.common.protocol.ApiKeys
-import org.apache.kafka.common.requests.ApiVersionsResponse
-import org.apache.kafka.network.SocketServerConfigs
-import org.apache.kafka.server.config.ServerConfigs
-import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotNull, assertTrue}
-import org.junit.jupiter.api.Timeout
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
-
-import java.io.{ByteArrayOutputStream, PrintStream}
-import java.nio.charset.StandardCharsets
-import java.util.Collections
-import scala.collection.Seq
-import scala.jdk.CollectionConverters._
-
-class BrokerApiVersionsCommandTest extends KafkaServerTestHarness {
-
- def generateConfigs: Seq[KafkaConfig] =
- if (isKRaftTest()) {
- TestUtils.createBrokerConfigs(1, null).map(props => {
- // Enable unstable api versions to be compatible with the new APIs under development,
- // maybe we can remove this after the new APIs is complete.
- props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true")
- props
- }).map(KafkaConfig.fromProps)
- } else {
- TestUtils.createBrokerConfigs(1, zkConnect).map(props => {
- // Configure control plane listener to make sure we have separate listeners from client,
- // in order to avoid returning Envelope API version.
- props.setProperty(SocketServerConfigs.CONTROL_PLANE_LISTENER_NAME_CONFIG, "CONTROLLER")
- props.setProperty(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT")
- props.setProperty("listeners", "PLAINTEXT://localhost:0,CONTROLLER://localhost:0")
- props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, "PLAINTEXT://localhost:0,CONTROLLER://localhost:0")
- props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true")
- props
- }).map(KafkaConfig.fromProps)
- }
-
- @Timeout(120)
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def checkBrokerApiVersionCommandOutput(quorum: String): Unit = {
- val byteArrayOutputStream = new ByteArrayOutputStream
- val printStream = new PrintStream(byteArrayOutputStream, false, StandardCharsets.UTF_8.name())
- BrokerApiVersionsCommand.execute(Array("--bootstrap-server", bootstrapServers()), printStream)
- val content = new String(byteArrayOutputStream.toByteArray, StandardCharsets.UTF_8)
- val lineIter = content.split("\n").iterator
- assertTrue(lineIter.hasNext)
- assertEquals(s"${bootstrapServers()} (id: 0 rack: null) -> (", lineIter.next())
- val listenerType = if (isKRaftTest()) {
- ApiMessageType.ListenerType.BROKER
- } else {
- ApiMessageType.ListenerType.ZK_BROKER
- }
- val clientApis = ApiKeys.clientApis().asScala
- val nodeApiVersions = new NodeApiVersions(clientApis.map(ApiVersionsResponse.toApiVersion).asJava, Collections.emptyList(), false)
- for (apiKey <- clientApis) {
- val terminator = if (apiKey == clientApis.last) "" else ","
- if (apiKey.inScope(listenerType)) {
- val apiVersion = nodeApiVersions.apiVersion(apiKey)
- assertNotNull(apiVersion)
-
- val versionRangeStr =
- if (apiVersion.minVersion == apiVersion.maxVersion) apiVersion.minVersion.toString
- else s"${apiVersion.minVersion} to ${apiVersion.maxVersion}"
- val usableVersion = nodeApiVersions.latestUsableVersion(apiKey)
-
- val line =
- if (apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey == ApiKeys.PUSH_TELEMETRY) s"\t${apiKey.name}(${apiKey.id}): UNSUPPORTED$terminator"
- else s"\t${apiKey.name}(${apiKey.id}): $versionRangeStr [usable: $usableVersion]$terminator"
- assertTrue(lineIter.hasNext)
- assertEquals(line, lineIter.next())
- } else {
- val line = s"\t${apiKey.name}(${apiKey.id}): UNSUPPORTED$terminator"
- assertTrue(lineIter.hasNext)
- assertEquals(line, lineIter.next())
- }
- }
- assertTrue(lineIter.hasNext)
- assertEquals(")", lineIter.next())
- assertFalse(lineIter.hasNext)
- }
-}
diff --git a/tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java b/tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java
new file mode 100644
index 00000000000..bf6a4e7adcf
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java
@@ -0,0 +1,296 @@
+/*
+ * 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 org.apache.kafka.tools;
+
+import org.apache.kafka.clients.ApiVersions;
+import org.apache.kafka.clients.ClientDnsLookup;
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.Metadata;
+import org.apache.kafka.clients.MetadataRecoveryStrategy;
+import org.apache.kafka.clients.NetworkClient;
+import org.apache.kafka.clients.NodeApiVersions;
+import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
+import org.apache.kafka.clients.consumer.internals.RequestFuture;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.errors.AuthenticationException;
+import org.apache.kafka.common.internals.ClusterResourceListeners;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.Selector;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ApiVersionsRequest;
+import org.apache.kafka.common.requests.ApiVersionsResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import joptsimple.OptionSpec;
+
+public class BrokerApiVersionsCommand {
+ public static void main(String... args) {
+ Exit.exit(mainNoExit(args));
+ }
+
+ static int mainNoExit(String... args) {
+ try {
+ execute(args);
+ return 0;
+ } catch (Throwable e) {
+ System.err.println(e.getMessage());
+ System.err.println(Utils.stackTrace(e));
+ return 1;
+ }
+ }
+
+ public static void execute(String... args) throws IOException, InterruptedException {
+ BrokerVersionCommandOptions opts = new BrokerVersionCommandOptions(args);
+ try (AdminClient adminClient = createAdminClient(opts)) {
+ adminClient.awaitBrokers();
+ Map<Node, KafkaFuture<NodeApiVersions>> brokerMap = adminClient.listAllBrokerVersionInfo();
+ brokerMap.forEach((broker, future) -> {
+ try {
+ NodeApiVersions apiVersions = future.get();
+ System.out.print(broker + " -> " + apiVersions.toString(true) + "\n");
+ } catch (Exception e) {
+ System.out.print(broker + " -> ERROR: " + e.getMessage() + "\n");
+ }
+ });
+ }
+ }
+
+ private static AdminClient createAdminClient(BrokerVersionCommandOptions opts) throws IOException {
+ Properties props = opts.options.has(opts.commandConfigOpt) ?
+ Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) :
+ new Properties();
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt));
+ return AdminClient.create(props);
+ }
+
+ private static class BrokerVersionCommandOptions extends CommandDefaultOptions {
+ private static final String BOOTSTRAP_SERVER_DOC = "REQUIRED: The server to connect to.";
+ private static final String COMMAND_CONFIG_DOC = "A property file containing configs to be passed to Admin Client.";
+
+ final OptionSpec<String> commandConfigOpt;
+ final OptionSpec<String> bootstrapServerOpt;
+
+ BrokerVersionCommandOptions(String[] args) {
+ super(args);
+ commandConfigOpt = parser.accepts("command-config", COMMAND_CONFIG_DOC)
+ .withRequiredArg()
+ .describedAs("command config property file")
+ .ofType(String.class);
+ bootstrapServerOpt = parser.accepts("bootstrap-server", BOOTSTRAP_SERVER_DOC)
+ .withRequiredArg()
+ .describedAs("server(s) to use for bootstrapping")
+ .ofType(String.class);
+ options = parser.parse(args);
+ checkArgs();
+ }
+
+ private void checkArgs() {
+ CommandLineUtils.maybePrintHelpOrVersion(this, "This tool helps to retrieve broker version information.");
+ CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt);
+ }
+ }
+
+ protected static class AdminClient implements AutoCloseable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(AdminClient.class);
+ private static final int DEFAULT_CONNECTION_MAX_IDLE_MS = 9 * 60 * 1000;
+ private static final int DEFAULT_REQUEST_TIMEOUT_MS = 5000;
+ private static final int DEFAULT_MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = 100;
+ private static final int DEFAULT_RECONNECT_BACKOFF_MS = 50;
+ private static final int DEFAULT_RECONNECT_BACKOFF_MAX = 50;
+ private static final int DEFAULT_SEND_BUFFER_BYTES = 128 * 1024;
+ private static final int DEFAULT_RECEIVE_BUFFER_BYTES = 32 * 1024;
+ private static final int DEFAULT_RETRY_BACKOFF_MS = 100;
+
+ private static final AtomicInteger ADMIN_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
+ private static final ConfigDef ADMIN_CONFIG_DEF = new ConfigDef()
+ .define(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, ConfigDef.Type.LIST, ConfigDef.Importance.HIGH, CommonClientConfigs.BOOTSTRAP_SERVERS_DOC)
+ .define(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG, ConfigDef.Type.STRING, ClientDnsLookup.USE_ALL_DNS_IPS.toString(), ConfigDef.ValidString.in(ClientDnsLookup.USE_ALL_DNS_IPS.toString(), ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()), ConfigDef.Importance.MEDIUM, CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
+ .define(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, ConfigDef.Type.STRING, CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL, ConfigDef.CaseInsensitiveValidString.in(Utils.enumOptions(SecurityProtocol.class)), ConfigDef.Importance.MEDIUM, CommonClientConfigs.SECURITY_PROTOCOL_DOC)
+ .define(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG, ConfigDef.Type.INT, DEFAULT_REQUEST_TIMEOUT_MS, ConfigDef.Importance.MEDIUM, CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC)
+ .define(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG, ConfigDef.Type.LONG, CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MS, ConfigDef.Importance.MEDIUM, CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_DOC)
+ .define(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG, ConfigDef.Type.LONG, CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS, ConfigDef.Importance.MEDIUM, CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_DOC)
+ .define(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG, ConfigDef.Type.LONG, DEFAULT_RETRY_BACKOFF_MS, ConfigDef.Importance.MEDIUM, CommonClientConfigs.RETRY_BACKOFF_MS_DOC)
+ .withClientSslSupport()
+ .withClientSaslSupport();
+
+ private final Time time;
+ private final ConsumerNetworkClient client;
+ private final List<Node> bootstrapBrokers;
+
+ static AdminClient create(Properties props) {
+ return create(new AbstractConfig(ADMIN_CONFIG_DEF, props, false));
+ }
+
+ static AdminClient create(AbstractConfig config) {
+ String clientId = "admin-" + ADMIN_CLIENT_ID_SEQUENCE.getAndIncrement();
+ LogContext logContext = new LogContext("[LegacyAdminClient clientId=" + clientId + "] ");
+ Time time = Time.SYSTEM;
+ Metrics metrics = new Metrics(time);
+ Metadata metadata = new Metadata(
+ CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MS,
+ CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MAX_MS,
+ 60 * 60 * 1000L, logContext,
+ new ClusterResourceListeners());
+ metadata.bootstrap(ClientUtils.parseAndValidateAddresses(
+ config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
+ config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG)));
+ Selector selector = new Selector(
+ DEFAULT_CONNECTION_MAX_IDLE_MS,
+ metrics,
+ time,
+ "admin",
+ ClientUtils.createChannelBuilder(config, time, logContext),
+ logContext);
+ NetworkClient networkClient = new NetworkClient(
+ selector,
+ metadata,
+ clientId,
+ DEFAULT_MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,
+ DEFAULT_RECONNECT_BACKOFF_MS,
+ DEFAULT_RECONNECT_BACKOFF_MAX,
+ DEFAULT_SEND_BUFFER_BYTES,
+ DEFAULT_RECEIVE_BUFFER_BYTES,
+ config.getInt(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG),
+ config.getLong(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG),
+ config.getLong(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG),
+ time,
+ true,
+ new ApiVersions(),
+ logContext,
+ MetadataRecoveryStrategy.NONE);
+ ConsumerNetworkClient highLevelClient = new ConsumerNetworkClient(
+ logContext,
+ networkClient,
+ metadata,
+ time,
+ config.getLong(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG),
+ config.getInt(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG),
+ Integer.MAX_VALUE);
+ return new AdminClient(time, highLevelClient, metadata.fetch().nodes());
+ }
+
+ AdminClient(Time time, ConsumerNetworkClient client, List<Node> bootstrapBrokers) {
+ this.time = time;
+ this.client = client;
+ this.bootstrapBrokers = bootstrapBrokers;
+ }
+
+ private AbstractResponse send(Node target, AbstractRequest.Builder<?> request) {
+ RequestFuture<ClientResponse> future = client.send(target, request);
+ while (!future.isDone()) {
+ client.poll(time.timer(DEFAULT_REQUEST_TIMEOUT_MS));
+ }
+ if (future.succeeded()) {
+ return future.value().responseBody();
+ } else {
+ throw future.exception();
+ }
+ }
+
+ private AbstractResponse sendAnyNode(AbstractRequest.Builder<?> request) {
+ for (Node broker : bootstrapBrokers) {
+ try {
+ return send(broker, request);
+ } catch (AuthenticationException e) {
+ throw e;
+ } catch (Exception e) {
+ LOGGER.debug("Request {} failed against node {}", request.apiKey(), broker, e);
+ }
+ }
+ throw new RuntimeException("Request " + request.apiKey() + " failed on brokers " + bootstrapBrokers);
+ }
+
+ protected KafkaFuture<NodeApiVersions> getNodeApiVersions(Node node) {
+ final KafkaFutureImpl<NodeApiVersions> future = new KafkaFutureImpl<>();
+ try {
+ ApiVersionsResponse response = (ApiVersionsResponse) send(node, new ApiVersionsRequest.Builder());
+ Errors error = Errors.forCode(response.data().errorCode());
+ if (error.exception() != null) {
+ future.completeExceptionally(error.exception());
+ } else {
+ future.complete(new NodeApiVersions(response.data().apiKeys(), response.data().supportedFeatures(), response.data().zkMigrationReady()));
+ }
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+
+ return future;
+ }
+
+ public void awaitBrokers() throws InterruptedException {
+ List<Node> nodes;
+ do {
+ nodes = findAllBrokers();
+ if (nodes.isEmpty()) {
+ TimeUnit.MILLISECONDS.sleep(50);
+ }
+ } while (nodes.isEmpty());
+ }
+
+ private List<Node> findAllBrokers() {
+ MetadataResponse response = (MetadataResponse) sendAnyNode(MetadataRequest.Builder.allTopics());
+ if (!response.errors().isEmpty()) {
+ LOGGER.debug("Metadata request contained errors: {}", response.errors());
+ }
+ return response.buildCluster().nodes();
+ }
+
+ public Map<Node, KafkaFuture<NodeApiVersions>> listAllBrokerVersionInfo() {
+ return findAllBrokers().stream()
+ .collect(Collectors.toMap(
+ broker -> broker,
+ this::getNodeApiVersions
+ ));
+ }
+
+ @Override
+ public void close() {
+ try {
+ client.close();
+ } catch (IOException e) {
+ LOGGER.error("Exception closing nioSelector:", e);
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/tools/src/test/java/org/apache/kafka/tools/BrokerApiVersionsCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/BrokerApiVersionsCommandTest.java
new file mode 100644
index 00000000000..4862de65cb0
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/BrokerApiVersionsCommandTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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 org.apache.kafka.tools;
+
+import kafka.test.ClusterInstance;
+import kafka.test.annotation.ClusterConfigProperty;
+import kafka.test.annotation.ClusterTest;
+import kafka.test.annotation.ClusterTestDefaults;
+import kafka.test.junit.ClusterTestExtensions;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.NodeApiVersions;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.message.ApiMessageType;
+import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.requests.ApiVersionsResponse;
+import org.apache.kafka.server.config.ServerConfigs;
+
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ClusterTestExtensions.class)
+@ClusterTestDefaults(serverProperties = {
+ @ClusterConfigProperty(key = ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, value = "true"),
+})
+public class BrokerApiVersionsCommandTest {
+ @ClusterTest
+ public void testBrokerApiVersionsCommandOutput(ClusterInstance clusterInstance) {
+ String output = ToolsTestUtils.grabConsoleOutput(() ->
+ BrokerApiVersionsCommand.mainNoExit("--bootstrap-server", clusterInstance.bootstrapServers()));
+ Iterator<String> lineIter = Arrays.stream(output.split("\n")).iterator();
+ assertTrue(lineIter.hasNext());
+ assertEquals(clusterInstance.bootstrapServers() + " (id: 0 rack: null) -> (", lineIter.next());
+
+ ApiMessageType.ListenerType listenerType = clusterInstance.isKRaftTest() ?
+ ApiMessageType.ListenerType.BROKER : ApiMessageType.ListenerType.ZK_BROKER;
+
+ NodeApiVersions nodeApiVersions = new NodeApiVersions(
+ ApiVersionsResponse.collectApis(ApiKeys.clientApis(), true),
+ Collections.emptyList(),
+ false);
+ Iterator<ApiKeys> apiKeysIter = ApiKeys.clientApis().iterator();
+ while (apiKeysIter.hasNext()) {
+ ApiKeys apiKey = apiKeysIter.next();
+ String terminator = apiKeysIter.hasNext() ? "," : "";
+ StringBuilder lineBuilder = new StringBuilder().append("\t");
+ if (apiKey.inScope(listenerType)) {
+ ApiVersion apiVersion = nodeApiVersions.apiVersion(apiKey);
+ assertNotNull(apiVersion);
+
+ String versionRangeStr = (apiVersion.minVersion() == apiVersion.maxVersion()) ?
+ String.valueOf(apiVersion.minVersion()) :
+ apiVersion.minVersion() + " to " + apiVersion.maxVersion();
+ short usableVersion = nodeApiVersions.latestUsableVersion(apiKey);
+ if (apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey == ApiKeys.PUSH_TELEMETRY) {
+ lineBuilder.append(apiKey.name).append("(").append(apiKey.id).append("): UNSUPPORTED").append(terminator);
+ } else {
+ lineBuilder.append(apiKey.name).append("(").append(apiKey.id).append("): ").append(versionRangeStr).append(" [usable: ").append(usableVersion).append("]").append(terminator);
+ }
+ } else {
+ lineBuilder.append(apiKey.name).append("(").append(apiKey.id).append("): UNSUPPORTED").append(terminator);
+ }
+ assertTrue(lineIter.hasNext());
+ assertEquals(lineBuilder.toString(), lineIter.next());
+ }
+ assertTrue(lineIter.hasNext());
+ assertEquals(")", lineIter.next());
+ assertFalse(lineIter.hasNext());
+ }
+
+ @ClusterTest
+ public void testAdminSendNoBlock(ClusterInstance clusterInstance) {
+ Properties props = new Properties();
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers());
+ try (BrokerApiVersionsCommand.AdminClient admin = BrokerApiVersionsCommand.AdminClient.create(props)) {
+ int brokerId = clusterInstance.brokers().keySet().iterator().next();
+ KafkaFuture<NodeApiVersions> future = admin.getNodeApiVersions(new Node(brokerId + 1, "localhost", 9093, null));
+ assertTrue(future.isCompletedExceptionally());
+ }
+ }
+}