KAFKA-15831: KIP-1000 protocol and admin client

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

@@ -1663,6 +1663,26 @@ default FenceProducersResult fenceProducers(Collection<String> transactionalIds)
     FenceProducersResult fenceProducers(Collection<String> transactionalIds,
                                         FenceProducersOptions options);
 
+    /**
+     * List the client metrics configuration resources available in the cluster.
+     *
+     * @param options The options to use when listing the client metrics resources.
+     * @return The ListClientMetricsResourcesResult.
+     */
+    ListClientMetricsResourcesResult listClientMetricsResources(ListClientMetricsResourcesOptions options);
+
+    /**
+     * List the client metrics configuration resources available in the cluster with the default options.
+     * <p>
+     * This is a convenience method for {@link #listClientMetricsResources(ListClientMetricsResourcesOptions)}
+     * with default options. See the overload for more details.
+     *
+     * @return The ListClientMetricsResourcesResult.
+     */
+    default ListClientMetricsResourcesResult listClientMetricsResources() {
+        return listClientMetricsResources(new ListClientMetricsResourcesOptions());
+    }
+
     /**
      * Determines the client's unique client instance ID used for telemetry. This ID is unique to
      * this specific client instance and will not change after it is initially generated.

clients/src/main/java/org/apache/kafka/clients/admin/ClientMetricsResourceListing.java

@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import java.util.Objects;
+
+@InterfaceStability.Evolving
+public class ClientMetricsResourceListing {
+    private final String name;
+
+    public ClientMetricsResourceListing(String name) {
+        this.name = name;
+    }
+
+    public String name() {
+        return name;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ClientMetricsResourceListing that = (ClientMetricsResourceListing) o;
+        return Objects.equals(name, that.name);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(name);
+    }
+
+    @Override
+    public String toString() {
+        return "ClientMetricsResourceListing(" +
+            "name='" + name +
+            ')';
+    }
+}

clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java

@@ -278,6 +278,11 @@ public FenceProducersResult fenceProducers(Collection<String> transactionalIds,
         return delegate.fenceProducers(transactionalIds, options);
     }
 
+    @Override
+    public ListClientMetricsResourcesResult listClientMetricsResources(ListClientMetricsResourcesOptions options) {
+        return delegate.listClientMetricsResources(options);
+    }
+
     @Override
     public Uuid clientInstanceId(Duration timeout) {
         return delegate.clientInstanceId(timeout);

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -141,6 +141,7 @@
 import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
 import org.apache.kafka.common.message.ExpireDelegationTokenRequestData;
 import org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity;
+import org.apache.kafka.common.message.ListClientMetricsResourcesRequestData;
 import org.apache.kafka.common.message.ListGroupsRequestData;
 import org.apache.kafka.common.message.ListGroupsResponseData;
 import org.apache.kafka.common.message.ListPartitionReassignmentsRequestData;
@@ -210,6 +211,8 @@
 import org.apache.kafka.common.requests.IncrementalAlterConfigsRequest;
 import org.apache.kafka.common.requests.IncrementalAlterConfigsResponse;
 import org.apache.kafka.common.requests.JoinGroupRequest;
+import org.apache.kafka.common.requests.ListClientMetricsResourcesRequest;
+import org.apache.kafka.common.requests.ListClientMetricsResourcesResponse;
 import org.apache.kafka.common.requests.ListGroupsRequest;
 import org.apache.kafka.common.requests.ListGroupsResponse;
 import org.apache.kafka.common.requests.ListOffsetsRequest;
@@ -4385,6 +4388,36 @@ public FenceProducersResult fenceProducers(Collection<String> transactionalIds,
         return new FenceProducersResult(future.all());
     }
 
+    @Override
+    public ListClientMetricsResourcesResult listClientMetricsResources(ListClientMetricsResourcesOptions options) {
+        final long now = time.milliseconds();
+        final KafkaFutureImpl<Collection<ClientMetricsResourceListing>> future = new KafkaFutureImpl<>();
+        runnable.call(new Call("listClientMetricsResources", calcDeadlineMs(now, options.timeoutMs()),
+            new LeastLoadedNodeProvider()) {
+
+            @Override
+            ListClientMetricsResourcesRequest.Builder createRequest(int timeoutMs) {
+                return new ListClientMetricsResourcesRequest.Builder(new ListClientMetricsResourcesRequestData());
+            }
+
+            @Override
+            void handleResponse(AbstractResponse abstractResponse) {
+                ListClientMetricsResourcesResponse response = (ListClientMetricsResourcesResponse) abstractResponse;
+                if (response.error().isFailure()) {
+                    future.completeExceptionally(response.error().exception());
+                } else {
+                    future.complete(response.clientMetricsResources());
+                }
+            }
+
+            @Override
+            void handleFailure(Throwable throwable) {
+                future.completeExceptionally(throwable);
+            }
+        }, now);
+        return new ListClientMetricsResourcesResult(future);
+    }
+
     @Override
     public Uuid clientInstanceId(Duration timeout) {
         throw new UnsupportedOperationException();

clients/src/main/java/org/apache/kafka/clients/admin/ListClientMetricsResourcesOptions.java

@@ -0,0 +1,29 @@
+/*
+ * 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 {@link Admin#listClientMetricsResources()}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListClientMetricsResourcesOptions extends AbstractOptions<ListClientMetricsResourcesOptions> {
+}

clients/src/main/java/org/apache/kafka/clients/admin/ListClientMetricsResourcesResult.java

@@ -0,0 +1,57 @@
+/*
+ * 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.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.Collection;
+
+/**
+ * The result of the {@link Admin#listClientMetricsResources()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListClientMetricsResourcesResult {
+    private final KafkaFuture<Collection<ClientMetricsResourceListing>> future;
+
+    ListClientMetricsResourcesResult(KafkaFuture<Collection<ClientMetricsResourceListing>> future) {
+        this.future = future;
+    }
+
+    /**
+     * Returns a future that yields either an exception, or the full set of client metrics
+     * listings.
+     *
+     * In the event of a failure, the future yields nothing but the first exception which
+     * occurred.
+     */
+    public KafkaFuture<Collection<ClientMetricsResourceListing>> all() {
+        final KafkaFutureImpl<Collection<ClientMetricsResourceListing>> result = new KafkaFutureImpl<>();
+        future.whenComplete((listings, throwable) -> {
+            if (throwable != null) {
+                result.completeExceptionally(throwable);
+            } else {
+                result.complete(listings);
+            }
+        });
+        return result;
+    }
+}

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -116,7 +116,8 @@ public enum ApiKeys {
     CONTROLLER_REGISTRATION(ApiMessageType.CONTROLLER_REGISTRATION),
     GET_TELEMETRY_SUBSCRIPTIONS(ApiMessageType.GET_TELEMETRY_SUBSCRIPTIONS),
     PUSH_TELEMETRY(ApiMessageType.PUSH_TELEMETRY),
-    ASSIGN_REPLICAS_TO_DIRS(ApiMessageType.ASSIGN_REPLICAS_TO_DIRS);
+    ASSIGN_REPLICAS_TO_DIRS(ApiMessageType.ASSIGN_REPLICAS_TO_DIRS),
+    LIST_CLIENT_METRICS_RESOURCES(ApiMessageType.LIST_CLIENT_METRICS_RESOURCES);
 
     private static final Map<ApiMessageType.ListenerType, EnumSet<ApiKeys>> APIS_BY_LISTENER =
         new EnumMap<>(ApiMessageType.ListenerType.class);

clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java

@@ -322,6 +322,8 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
                 return PushTelemetryRequest.parse(buffer, apiVersion);
             case ASSIGN_REPLICAS_TO_DIRS:
                 return AssignReplicasToDirsRequest.parse(buffer, apiVersion);
+            case LIST_CLIENT_METRICS_RESOURCES:
+                return ListClientMetricsResourcesRequest.parse(buffer, apiVersion);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java

@@ -259,6 +259,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, ByteBuffer response
                 return PushTelemetryResponse.parse(responseBuffer, version);
             case ASSIGN_REPLICAS_TO_DIRS:
                 return AssignReplicasToDirsResponse.parse(responseBuffer, version);
+            case LIST_CLIENT_METRICS_RESOURCES:
+                return ListClientMetricsResourcesResponse.parse(responseBuffer, version);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/ListClientMetricsResourcesRequest.java

@@ -0,0 +1,77 @@
+/*
+ * 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.common.requests;
+
+import org.apache.kafka.common.message.ListClientMetricsResourcesRequestData;
+import org.apache.kafka.common.message.ListClientMetricsResourcesResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+
+import java.nio.ByteBuffer;
+
+public class ListClientMetricsResourcesRequest extends AbstractRequest {
+    public static class Builder extends AbstractRequest.Builder<ListClientMetricsResourcesRequest> {
+        public final ListClientMetricsResourcesRequestData data;
+
+        public Builder(ListClientMetricsResourcesRequestData data) {
+            super(ApiKeys.LIST_CLIENT_METRICS_RESOURCES);
+            this.data = data;
+        }
+
+        @Override
+        public ListClientMetricsResourcesRequest build(short version) {
+            return new ListClientMetricsResourcesRequest(data, version);
+        }
+
+        @Override
+        public String toString() {
+            return data.toString();
+        }
+    }
+
+    private final ListClientMetricsResourcesRequestData data;
+
+    private ListClientMetricsResourcesRequest(ListClientMetricsResourcesRequestData data, short version) {
+        super(ApiKeys.LIST_CLIENT_METRICS_RESOURCES, version);
+        this.data = data;
+    }
+
+    public ListClientMetricsResourcesRequestData data() {
+        return data;
+    }
+
+    @Override
+    public ListClientMetricsResourcesResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        Errors error = Errors.forException(e);
+        ListClientMetricsResourcesResponseData response = new ListClientMetricsResourcesResponseData()
+            .setErrorCode(error.code())
+            .setThrottleTimeMs(throttleTimeMs);
+        return new ListClientMetricsResourcesResponse(response);
+    }
+
+    public static ListClientMetricsResourcesRequest parse(ByteBuffer buffer, short version) {
+        return new ListClientMetricsResourcesRequest(new ListClientMetricsResourcesRequestData(
+            new ByteBufferAccessor(buffer), version), version);
+    }
+
+    @Override
+    public String toString(boolean verbose) {
+        return data.toString();
+    }
+
+}