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 3b849ff2bdf KAFKA-19663 Unify kafka-broker-api-versions tool to
cluster-tool and support bootstrap controllers (#20598)
3b849ff2bdf is described below
commit 3b849ff2bdf245722b462d5f3c92f47088fb772a
Author: TaiJuWu <[email protected]>
AuthorDate: Thu Jul 16 19:37:57 2026 +0800
KAFKA-19663 Unify kafka-broker-api-versions tool to cluster-tool and
support bootstrap controllers (#20598)
Unify kafka-broker-api-versions tool to cluster-tool and support
bootstrap controllers. This change reuse current KafkaAdminClient
describeFeature to get api response by introducing
InternalFeatureDescribeFeatureResult.
Reviewers: Chia-Ping Tsai <[email protected]>
---
bin/kafka-broker-api-versions.sh | 1 +
bin/windows/kafka-broker-api-versions.bat | 1 +
.../kafka/clients/admin/DescribeFeaturesTest.java | 16 ++++
.../clients/admin/DescribeFeaturesResult.java | 8 +-
.../kafka/clients/admin/KafkaAdminClient.java | 18 ++++-
.../internals/InternalDescribeFeaturesResult.java | 47 ++++++++++++
docs/getting-started/upgrade.md | 1 +
.../java/org/apache/kafka/tools/ClusterTool.java | 38 +++++++++-
.../org/apache/kafka/tools/ClusterToolTest.java | 85 ++++++++++++++++++++++
9 files changed, 211 insertions(+), 4 deletions(-)
diff --git a/bin/kafka-broker-api-versions.sh b/bin/kafka-broker-api-versions.sh
index 285a9488af0..8659a17d290 100755
--- a/bin/kafka-broker-api-versions.sh
+++ b/bin/kafka-broker-api-versions.sh
@@ -14,4 +14,5 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+echo "WARNING: kafka-broker-api-versions.sh is deprecated and will be removed
in the next major release. Use 'kafka-cluster.sh api-versions' instead." 1>&2
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 7c0444bc06f..ba65e9bcde2 100644
--- a/bin/windows/kafka-broker-api-versions.bat
+++ b/bin/windows/kafka-broker-api-versions.bat
@@ -14,4 +14,5 @@ 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.
+echo WARNING: kafka-broker-api-versions.bat is deprecated and will be removed
in the next major release. Use "kafka-cluster.bat api-versions" instead. 1>&2
%~dp0kafka-run-class.bat org.apache.kafka.tools.BrokerApiVersionsCommand %*
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeFeaturesTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeFeaturesTest.java
index 6d1fe1903cc..1f03e8091d2 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeFeaturesTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeFeaturesTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.clients.admin;
+import org.apache.kafka.clients.admin.internals.InternalDescribeFeaturesResult;
+import org.apache.kafka.common.message.ApiMessageType;
import org.apache.kafka.common.test.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
@@ -35,9 +37,23 @@ import java.util.concurrent.ExecutionException;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class DescribeFeaturesTest {
+ @ClusterTest(types = {Type.KRAFT})
+ public void testApiVersions(ClusterInstance clusterInstance) throws
ExecutionException, InterruptedException {
+ try (Admin admin = clusterInstance.admin()) {
+ var versions = ((InternalDescribeFeaturesResult)
admin.describeFeatures()).nodeApiVersions().get();
+ versions.allSupportedApiVersions().forEach((key, version) ->
assertTrue(key.inScope(ApiMessageType.ListenerType.BROKER)));
+ }
+
+ try (Admin admin = clusterInstance.admin(Map.of(), true)) {
+ var versions = ((InternalDescribeFeaturesResult)
admin.describeFeatures()).nodeApiVersions().get();
+ versions.allSupportedApiVersions().forEach((key, version) ->
assertTrue(key.inScope(ApiMessageType.ListenerType.CONTROLLER)));
+ }
+ }
+
@ClusterTest(
types = {Type.KRAFT},
metadataVersion = MetadataVersion.IBP_3_7_IV0,
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeFeaturesResult.java
b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeFeaturesResult.java
index 2b8c3fa521e..edd5d365249 100644
---
a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeFeaturesResult.java
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeFeaturesResult.java
@@ -30,7 +30,13 @@ public class DescribeFeaturesResult {
private final KafkaFuture<FeatureMetadata> future;
- DescribeFeaturesResult(KafkaFuture<FeatureMetadata> future) {
+ /**
+ * This constructor is {@code protected} only to allow internal subclasses
(e.g.
+ * {@link
org.apache.kafka.clients.admin.internals.InternalDescribeFeaturesResult}) to
reuse it.
+ * It is not part of the public API contract, so binary/source
compatibility for subclassing
+ * outside of the Kafka clients module is not guaranteed.
+ */
+ protected DescribeFeaturesResult(KafkaFuture<FeatureMetadata> future) {
this.future = future;
}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 8f2dcc0e710..6392a1706f2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -28,6 +28,7 @@ import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.LeastLoadedNode;
import org.apache.kafka.clients.MetadataRecoveryStrategy;
import org.apache.kafka.clients.NetworkClient;
+import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.StaleMetadataException;
import
org.apache.kafka.clients.admin.CreateTopicsResult.TopicMetadataAndConfig;
import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResult;
@@ -59,6 +60,7 @@ import
org.apache.kafka.clients.admin.internals.DescribeShareGroupsHandler;
import org.apache.kafka.clients.admin.internals.DescribeStreamsGroupsHandler;
import org.apache.kafka.clients.admin.internals.DescribeTransactionsHandler;
import org.apache.kafka.clients.admin.internals.FenceProducersHandler;
+import org.apache.kafka.clients.admin.internals.InternalDescribeFeaturesResult;
import
org.apache.kafka.clients.admin.internals.ListConsumerGroupOffsetsHandler;
import org.apache.kafka.clients.admin.internals.ListOffsetsHandler;
import org.apache.kafka.clients.admin.internals.ListShareGroupOffsetsHandler;
@@ -4548,12 +4550,21 @@ public class KafkaAdminClient extends AdminClient {
@Override
public DescribeFeaturesResult describeFeatures(final
DescribeFeaturesOptions options) {
final KafkaFutureImpl<FeatureMetadata> future = new
KafkaFutureImpl<>();
+ final KafkaFutureImpl<NodeApiVersions> nodeApiVersionsFuture = new
KafkaFutureImpl<>();
final long now = time.milliseconds();
final NodeProvider nodeProvider = options.nodeId().isPresent() ?
new ConstantNodeIdProvider(options.nodeId().getAsInt(), true) :
new LeastLoadedBrokerOrActiveKController();
final Call call = new Call(
"describeFeatures", calcDeadlineMs(now, options.timeoutMs()),
nodeProvider) {
+ private NodeApiVersions createNodeApiVersion(final
ApiVersionsResponse response) {
+ return new NodeApiVersions(
+ response.data().apiKeys(),
+ response.data().supportedFeatures(),
+ response.data().finalizedFeatures(),
+ response.data().finalizedFeaturesEpoch());
+ }
+
private FeatureMetadata createFeatureMetadata(final
ApiVersionsResponse response) {
final Map<String, FinalizedVersionRange> finalizedFeatures =
new HashMap<>();
for (final FinalizedFeatureKey key :
response.data().finalizedFeatures().valuesSet()) {
@@ -4585,8 +4596,11 @@ public class KafkaAdminClient extends AdminClient {
final ApiVersionsResponse apiVersionsResponse =
(ApiVersionsResponse) response;
if (apiVersionsResponse.data().errorCode() ==
Errors.NONE.code()) {
future.complete(createFeatureMetadata(apiVersionsResponse));
+
nodeApiVersionsFuture.complete(createNodeApiVersion(apiVersionsResponse));
} else {
-
future.completeExceptionally(Errors.forCode(apiVersionsResponse.data().errorCode()).exception());
+ Exception exception =
Errors.forCode(apiVersionsResponse.data().errorCode()).exception();
+ future.completeExceptionally(exception);
+ nodeApiVersionsFuture.completeExceptionally(exception);
}
}
@@ -4597,7 +4611,7 @@ public class KafkaAdminClient extends AdminClient {
};
runnable.call(call, now);
- return new DescribeFeaturesResult(future);
+ return new InternalDescribeFeaturesResult(future,
nodeApiVersionsFuture);
}
@Override
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/internals/InternalDescribeFeaturesResult.java
b/clients/src/main/java/org/apache/kafka/clients/admin/internals/InternalDescribeFeaturesResult.java
new file mode 100644
index 00000000000..b7e6cd262d3
--- /dev/null
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/internals/InternalDescribeFeaturesResult.java
@@ -0,0 +1,47 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.NodeApiVersions;
+import org.apache.kafka.clients.admin.DescribeFeaturesResult;
+import org.apache.kafka.clients.admin.FeatureMetadata;
+import org.apache.kafka.common.KafkaFuture;
+
+/**
+ * Internal result class for describeFeatures that exposes API version
information.
+ * This class is intended for use by internal Kafka tools that need access to
the raw API versions
+ * returned in the ApiVersionsResponse.
+ */
+public class InternalDescribeFeaturesResult extends DescribeFeaturesResult {
+
+ private final KafkaFuture<NodeApiVersions> nodeApiVersions;
+
+ public InternalDescribeFeaturesResult(KafkaFuture<FeatureMetadata>
featureMetadata, KafkaFuture<NodeApiVersions> nodeApiVersions) {
+ super(featureMetadata);
+ this.nodeApiVersions = nodeApiVersions;
+ }
+
+ /**
+ * Returns the node API versions future. This contains the API keys and
version ranges
+ * supported by the broker, as returned in the ApiVersionsResponse.
+ *
+ * @return KafkaFuture containing the node API versions
+ */
+ public KafkaFuture<NodeApiVersions> nodeApiVersions() {
+ return nodeApiVersions;
+ }
+}
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
index f960fc36235..ca6ab9923db 100644
--- a/docs/getting-started/upgrade.md
+++ b/docs/getting-started/upgrade.md
@@ -44,6 +44,7 @@ type: docs
For further details, please refer to
[KIP-1301](https://cwiki.apache.org/confluence/x/Z5U8G).
* The broker-side OAUTHBEARER JWT validator now fails fast at startup when a
JWKS endpoint (`sasl.oauthbearer.jwks.endpoint.url`) is configured but
`sasl.oauthbearer.expected.audience` or `sasl.oauthbearer.expected.issuer` is
not set. Brokers that previously started without these settings will now fail
to start until they are configured. To intentionally accept tokens regardless
of their audience or issuer, set the new
`sasl.oauthbearer.allow.unverified.audience` or `sasl.oauthbearer.a [...]
* When clients connect to the cluster, they now include cluster and node
information to enable detection and handling of misrouted connections. For
further details, please refer to
[KIP-1242](https://cwiki.apache.org/confluence/x/W4LMFw).
+ * The `kafka-cluster.sh` tool now provides an `api-versions` command to
display the API versions supported by the brokers or controllers, and it
accepts both `--bootstrap-server` and `--bootstrap-controller`. As a result,
`kafka-broker-api-versions.sh` is deprecated and will be removed in the next
major release; use `kafka-cluster.sh api-versions` instead. For further
details, please refer to
[KIP-1220](https://cwiki.apache.org/confluence/x/-QkbFw).
* Brokers can now record a human-readable description of each streams
group's processing topology via a pluggable backend, retrievable through
`Admin#describeStreamsGroups` and `kafka-streams-groups.sh --describe
--topology`. The feature is disabled unless the new broker configuration
`group.streams.topology.description.plugin.class` is set to a
`StreamsGroupTopologyDescriptionPlugin` implementation; on the client side, the
new Kafka Streams configuration `topology.description.push.ena [...]
## Upgrading to 4.3.0
diff --git a/tools/src/main/java/org/apache/kafka/tools/ClusterTool.java
b/tools/src/main/java/org/apache/kafka/tools/ClusterTool.java
index 96346874f9c..4342bb07bf3 100644
--- a/tools/src/main/java/org/apache/kafka/tools/ClusterTool.java
+++ b/tools/src/main/java/org/apache/kafka/tools/ClusterTool.java
@@ -16,8 +16,11 @@
*/
package org.apache.kafka.tools;
+import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.DescribeClusterOptions;
+import org.apache.kafka.clients.admin.DescribeFeaturesOptions;
+import org.apache.kafka.clients.admin.internals.InternalDescribeFeaturesResult;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.utils.Utils;
@@ -34,9 +37,12 @@ import net.sourceforge.argparse4j.inf.Subparsers;
import java.io.PrintStream;
import java.util.Collection;
+import java.util.Comparator;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Properties;
+import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import static net.sourceforge.argparse4j.impl.Arguments.store;
@@ -75,7 +81,9 @@ public class ClusterTool {
.help("Unregister a broker.");
Subparser listEndpoints = subparsers.addParser("list-endpoints")
.help("List endpoints");
- for (Subparser subpparser : List.of(clusterIdParser, unregisterParser,
listEndpoints)) {
+ Subparser apiVersionsParser = subparsers.addParser("api-versions")
+ .help("Get information about the api versions of the brokers
or controllers.");
+ for (Subparser subpparser : List.of(clusterIdParser, unregisterParser,
listEndpoints, apiVersionsParser)) {
MutuallyExclusiveGroup connectionOptions =
subpparser.addMutuallyExclusiveGroup().required(true);
connectionOptions.addArgument("--bootstrap-server", "-b")
.action(store())
@@ -141,6 +149,12 @@ public class ClusterTool {
}
break;
}
+ case "api-versions": {
+ try (Admin adminClient = Admin.create(properties)) {
+ apiVersionsCommand(System.out, adminClient);
+ }
+ break;
+ }
default:
throw new RuntimeException("Unknown command " + command);
}
@@ -208,4 +222,26 @@ public class ClusterTool {
}
}
}
+
+ static void apiVersionsCommand(PrintStream stream, Admin adminClient)
throws Exception {
+ Collection<Node> nodes = adminClient.describeCluster().nodes().get();
+ Map<Node, InternalDescribeFeaturesResult> nodeApiVersions = new
TreeMap<>(Comparator.comparingInt(Node::id));
+ nodes.forEach(node -> {
+ InternalDescribeFeaturesResult result =
(InternalDescribeFeaturesResult) adminClient.describeFeatures(
+ new DescribeFeaturesOptions().nodeId(node.id()));
+ nodeApiVersions.put(node, result);
+ });
+
+ nodeApiVersions.forEach((broker, future) -> {
+ try {
+ NodeApiVersions apiVersions = future.nodeApiVersions().get();
+ stream.print(broker + " -> " + apiVersions.toString(true) +
"\n");
+ } catch (ExecutionException e) {
+ stream.print(broker + " -> ERROR: " + e.getCause() + "\n");
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ stream.print(broker + " -> ERROR: " + e + "\n");
+ }
+ });
+ }
}
diff --git a/tools/src/test/java/org/apache/kafka/tools/ClusterToolTest.java
b/tools/src/test/java/org/apache/kafka/tools/ClusterToolTest.java
index 6207cc740fd..bbd5f2d1c23 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ClusterToolTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ClusterToolTest.java
@@ -16,12 +16,19 @@
*/
package org.apache.kafka.tools;
+import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.MockAdminClient;
+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.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.test.TestUtils;
import net.sourceforge.argparse4j.inf.ArgumentParserException;
@@ -36,12 +43,16 @@ import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
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.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -187,6 +198,80 @@ public class ClusterToolTest {
assertEquals("--config and --command-config cannot be specified
together.", ex.getMessage());
}
+ @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, serverProperties = {
+ @ClusterConfigProperty(key =
ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, value = "true"),
+ })
+ public void testApiVersionsCommandOutputUsingBrokerServer(ClusterInstance
clusterInstance) {
+ testApiVersionsCommandOutput(clusterInstance, false);
+ }
+
+ @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, serverProperties = {
+ @ClusterConfigProperty(key =
ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, value = "true"),
+ })
+ public void testApiVersionsCommandOutputUsingController(ClusterInstance
clusterInstance) {
+ testApiVersionsCommandOutput(clusterInstance, true);
+ }
+
+ private void testApiVersionsCommandOutput(ClusterInstance clusterInstance,
boolean usingBootstrapController) {
+ String output = ToolsTestUtils.grabConsoleOutput(() -> {
+ if (usingBootstrapController) {
+ assertDoesNotThrow(() -> ClusterTool.execute("api-versions",
"--bootstrap-controller", clusterInstance.bootstrapControllers()));
+ } else {
+ assertDoesNotThrow(() -> ClusterTool.execute("api-versions",
"--bootstrap-server", clusterInstance.bootstrapServers()));
+ }
+ });
+ Iterator<String> lineIter =
Arrays.stream(output.split("\n")).iterator();
+ assertTrue(lineIter.hasNext());
+ ApiMessageType.ListenerType listenerType = usingBootstrapController ?
+ ApiMessageType.ListenerType.CONTROLLER :
ApiMessageType.ListenerType.BROKER;
+
+ if (usingBootstrapController) {
+ int id = clusterInstance.type() == Type.CO_KRAFT ? 0 : 3000;
+ assertEquals(clusterInstance.bootstrapControllers() + " (id: " +
id + " rack: null isFenced: false) -> (", lineIter.next());
+ } else {
+ assertEquals(clusterInstance.bootstrapServers() + " (id: 0 rack:
null isFenced: false) -> (", lineIter.next());
+ }
+
+ EnumSet<ApiKeys> apiKeys = EnumSet.copyOf(ApiKeys.clientApis());
+
+ // Controller will return all apis
+ if (listenerType == ApiMessageType.ListenerType.CONTROLLER) {
+ apiKeys.addAll(ApiKeys.controllerApis());
+ }
+
+ Iterator<ApiKeys> apiKeysIter = apiKeys.iterator();
+
+ NodeApiVersions nodeApiVersions = new NodeApiVersions(
+ ApiVersionsResponse.filterApis(listenerType, true, true),
+ List.of());
+ 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, "No apiVersion found for " + apiKey);
+
+ 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());
+ }
+
@Test
public void testPrintClusterId() throws Exception {
Admin adminClient = new MockAdminClient.Builder().
