You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2017/06/14 01:05:08 UTC
kafka git commit: MINOR: Temporarily remove the apiVersions API for
0.11
Repository: kafka
Updated Branches:
refs/heads/trunk d099c7495 -> 32f0f1f50
MINOR: Temporarily remove the apiVersions API for 0.11
Author: Colin P. Mccabe <cm...@confluent.io>
Reviewers: Ismael Juma <is...@juma.me.uk>
Closes #3324 from cmccabe/removeApiVersions
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/32f0f1f5
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/32f0f1f5
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/32f0f1f5
Branch: refs/heads/trunk
Commit: 32f0f1f50aa618e8a3c52eb37006f4945fa6865a
Parents: d099c74
Author: Colin P. Mccabe <cm...@confluent.io>
Authored: Wed Jun 14 02:05:02 2017 +0100
Committer: Ismael Juma <is...@juma.me.uk>
Committed: Wed Jun 14 02:05:02 2017 +0100
----------------------------------------------------------------------
.../apache/kafka/clients/admin/AdminClient.java | 21 -------
.../kafka/clients/admin/ApiVersionsOptions.java | 37 ------------
.../kafka/clients/admin/ApiVersionsResult.java | 63 --------------------
.../kafka/clients/admin/KafkaAdminClient.java | 35 -----------
.../kafka/api/AdminClientIntegrationTest.scala | 10 ++--
5 files changed, 4 insertions(+), 162 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/32f0f1f5/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
index 7e5944f..a6c1086 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
@@ -17,7 +17,6 @@
package org.apache.kafka.clients.admin;
-import org.apache.kafka.common.Node;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.annotation.InterfaceStability;
@@ -193,26 +192,6 @@ public abstract class AdminClient implements AutoCloseable {
public abstract DescribeClusterResult describeCluster(DescribeClusterOptions options);
/**
- * Get information about the api versions of nodes in the cluster with the default options.
- * See {@link AdminClient#apiVersions(Collection<Node>, ApiVersionsOptions)}
- *
- * @param nodes The nodes to get information about, or null to get information about all nodes.
- * @return The ApiVersionsResult.
- */
- public ApiVersionsResult apiVersions(Collection<Node> nodes) {
- return apiVersions(nodes, new ApiVersionsOptions());
- }
-
- /**
- * Get information about the api versions of nodes in the cluster.
- *
- * @param nodes The nodes to get information about, or null to get information about all nodes.
- * @param options The options to use when getting api versions of the nodes.
- * @return The ApiVersionsResult.
- */
- public abstract ApiVersionsResult apiVersions(Collection<Node> nodes, ApiVersionsOptions options);
-
- /**
* Similar to #{@link AdminClient#describeAcls(AclBindingFilter, DescribeAclsOptions)},
* but uses the default options.
*
http://git-wip-us.apache.org/repos/asf/kafka/blob/32f0f1f5/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsOptions.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsOptions.java
deleted file mode 100644
index cbcd234..0000000
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsOptions.java
+++ /dev/null
@@ -1,37 +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 org.apache.kafka.clients.admin;
-
-import org.apache.kafka.common.annotation.InterfaceStability;
-
-/**
- * Options for the apiVersions call.
- */
-@InterfaceStability.Unstable
-public class ApiVersionsOptions {
- private Integer timeoutMs = null;
-
- public ApiVersionsOptions timeoutMs(Integer timeoutMs) {
- this.timeoutMs = timeoutMs;
- return this;
- }
-
- public Integer timeoutMs() {
- return timeoutMs;
- }
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/32f0f1f5/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsResult.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsResult.java
deleted file mode 100644
index 62b6d7f..0000000
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ApiVersionsResult.java
+++ /dev/null
@@ -1,63 +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 org.apache.kafka.clients.admin;
-
-import org.apache.kafka.clients.NodeApiVersions;
-import org.apache.kafka.common.KafkaFuture;
-import org.apache.kafka.common.Node;
-import org.apache.kafka.common.annotation.InterfaceStability;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-
-/**
- * Results of the apiVersions call.
- */
-@InterfaceStability.Unstable
-public class ApiVersionsResult {
- private final Map<Node, KafkaFuture<NodeApiVersions>> futures;
-
- ApiVersionsResult(Map<Node, KafkaFuture<NodeApiVersions>> futures) {
- this.futures = futures;
- }
-
- public Map<Node, KafkaFuture<NodeApiVersions>> results() {
- return futures;
- }
-
- public KafkaFuture<Map<Node, NodeApiVersions>> all() {
- return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).
- thenApply(new KafkaFuture.Function<Void, Map<Node, NodeApiVersions>>() {
- @Override
- public Map<Node, NodeApiVersions> apply(Void v) {
- Map<Node, NodeApiVersions> versions = new HashMap<>(futures.size());
- for (Map.Entry<Node, KafkaFuture<NodeApiVersions>> entry : futures.entrySet()) {
- try {
- versions.put(entry.getKey(), entry.getValue().get());
- } catch (InterruptedException | ExecutionException e) {
- // This should be unreachable, because allOf ensured that all the futures
- // completed successfully.
- throw new RuntimeException(e);
- }
- }
- return versions;
- }
- });
- }
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/32f0f1f5/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
----------------------------------------------------------------------
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 62899a3..849ff58 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
@@ -24,7 +24,6 @@ import org.apache.kafka.clients.ClientUtils;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.NetworkClient;
-import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResult;
import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResults;
import org.apache.kafka.common.Cluster;
@@ -59,8 +58,6 @@ import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.AlterConfigsRequest;
import org.apache.kafka.common.requests.AlterConfigsResponse;
-import org.apache.kafka.common.requests.ApiVersionsRequest;
-import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.requests.CreateAclsRequest;
import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation;
import org.apache.kafka.common.requests.CreateAclsResponse;
@@ -1264,38 +1261,6 @@ public class KafkaAdminClient extends AdminClient {
}
@Override
- public ApiVersionsResult apiVersions(Collection<Node> nodes, ApiVersionsOptions options) {
- final long now = time.milliseconds();
- final long deadlineMs = calcDeadlineMs(now, options.timeoutMs());
- Map<Node, KafkaFuture<NodeApiVersions>> nodeFutures = new HashMap<>();
- for (final Node node : nodes) {
- if (nodeFutures.get(node) != null)
- continue;
- final KafkaFutureImpl<NodeApiVersions> nodeFuture = new KafkaFutureImpl<>();
- nodeFutures.put(node, nodeFuture);
- runnable.call(new Call("apiVersions", deadlineMs, new ConstantNodeProvider(node)) {
- @Override
- public AbstractRequest.Builder createRequest(int timeoutMs) {
- return new ApiVersionsRequest.Builder();
- }
-
- @Override
- public void handleResponse(AbstractResponse abstractResponse) {
- ApiVersionsResponse response = (ApiVersionsResponse) abstractResponse;
- nodeFuture.complete(new NodeApiVersions(response.apiVersions()));
- }
-
- @Override
- public void handleFailure(Throwable throwable) {
- nodeFuture.completeExceptionally(throwable);
- }
- }, now);
- }
- return new ApiVersionsResult(nodeFutures);
-
- }
-
- @Override
public DescribeAclsResult describeAcls(final AclBindingFilter filter, DescribeAclsOptions options) {
final long now = time.milliseconds();
final KafkaFutureImpl<Collection<AclBinding>> future = new KafkaFutureImpl<>();
http://git-wip-us.apache.org/repos/asf/kafka/blob/32f0f1f5/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index e91c171..5e94ffe 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -199,21 +199,19 @@ class AdminClientIntegrationTest extends KafkaServerTestHarness with Logging {
}
@Test
- def testGetAllBrokerVersionsAndDescribeCluster(): Unit = {
+ def testDescribeCluster(): Unit = {
client = AdminClient.create(createConfig())
- val nodes = client.describeCluster().nodes().get()
+ val nodes = client.describeCluster.nodes.get()
val clusterId = client.describeCluster().clusterId().get()
assertEquals(servers.head.apis.clusterId, clusterId)
val controller = client.describeCluster().controller().get()
assertEquals(servers.head.apis.metadataCache.getControllerId.
getOrElse(MetadataResponse.NO_CONTROLLER_ID), controller.id())
- val nodesToVersions = client.apiVersions(nodes).all().get()
val brokers = brokerList.split(",")
- assert(brokers.size == nodesToVersions.size())
- for ((node, brokerVersionInfo) <- nodesToVersions.asScala) {
+ assertEquals(brokers.size, nodes.size)
+ for (node <- nodes.asScala) {
val hostStr = s"${node.host}:${node.port}"
assertTrue(s"Unknown host:port pair $hostStr in brokerVersionInfos", brokers.contains(hostStr))
- assertEquals(1, brokerVersionInfo.usableVersion(ApiKeys.API_VERSIONS))
}
}