KIP-546 (1/2): Implements describeClientQuotas() and alterClientQuotas().

build.gradle

@@ -457,7 +457,7 @@ subprojects {
     // See https://www.lightbend.com/blog/scala-inliner-optimizer for more information about the optimizer.
     scalaCompileOptions.additionalParameters += ["-opt:l:inline"]
     scalaCompileOptions.additionalParameters += inlineFrom
-    
+
     // these options are valid for Scala versions < 2.13 only
     // Scala 2.13 removes them, see https://github.com/scala/scala/pull/6502 and https://github.com/scala/scala/pull/5969
     if (versions.baseScala == '2.12') {

checkstyle/import-control.xml

@@ -145,6 +145,7 @@
       <allow pkg="org.apache.kafka.common.protocol" />
       <allow pkg="org.apache.kafka.common.message" />
       <allow pkg="org.apache.kafka.common.network" />
+      <allow pkg="org.apache.kafka.common.quota" />
       <allow pkg="org.apache.kafka.common.requests" />
       <allow pkg="org.apache.kafka.common.resource" />
       <allow pkg="org.apache.kafka.common.record" />
@@ -162,6 +163,10 @@
     <subpackage name="utils">
       <allow pkg="org.apache.kafka.common" />
     </subpackage>
+
+    <subpackage name="quotas">
+      <allow pkg="org.apache.kafka.common" />
+    </subpackage>
   </subpackage>
 
   <subpackage name="clients">
@@ -241,7 +246,7 @@
     <subpackage name="perf">
       <allow pkg="com.fasterxml.jackson.databind" />
     </subpackage>
-    
+
     <subpackage name="integration">
       <allow pkg="kafka.admin" />
       <allow pkg="kafka.api" />

checkstyle/suppressions.xml

@@ -62,7 +62,7 @@
               files="(ConsumerCoordinator|Fetcher|Sender|KafkaProducer|BufferPool|ConfigDef|RecordAccumulator|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager).java"/>
 
     <suppress checks="JavaNCSS"
-              files="(AbstractRequest|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest).java"/>
+              files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest).java"/>
 
     <suppress checks="NPathComplexity"
               files="(BufferPool|Fetcher|MetricName|Node|ConfigDef|RecordBatch|SslFactory|SslTransportLayer|MetadataResponse|KerberosLogin|Selector|Sender|Serdes|TokenInformation|Agent|Values|PluginUtils|MiniTrogdorCluster|TasksRequest|KafkaProducer).java"/>

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

@@ -36,6 +36,8 @@
 import org.apache.kafka.common.acl.AclBindingFilter;
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.quota.ClientQuotaAlteration;
+import org.apache.kafka.common.quota.ClientQuotaFilter;
 import org.apache.kafka.common.requests.LeaveGroupResponse;
 
 /**
@@ -408,7 +410,6 @@ default AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collectio
         return incrementalAlterConfigs(configs, new AlterConfigsOptions());
     }
 
-
     /**
      * Incrementally update the configuration for the specified resources.
      * <p>
@@ -1132,6 +1133,85 @@ default ListOffsetsResult listOffsets(Map<TopicPartition, OffsetSpec> topicParti
      */
     ListOffsetsResult listOffsets(Map<TopicPartition, OffsetSpec> topicPartitionOffsets, ListOffsetsOptions options);
 
+    /**
+     * Describes all entities matching the provided filter that have at least one client quota configuration
+     * value defined.
+     * <p>
+     * This is a convenience method for {@link #describeClientQuotas(ClientQuotaFilter, DescribeClientQuotasOptions)}
+     * with default options. See the overload for more details.
+     * <p>
+     * This operation is supported by brokers with version 2.6.0 or higher.
+     *
+     * @param filter the filter to apply to match entities
+     * @return the DescribeClientQuotasResult containing the result
+     */
+    default DescribeClientQuotasResult describeClientQuotas(ClientQuotaFilter filter) {
+        return describeClientQuotas(filter, new DescribeClientQuotasOptions());
+    }
+
+    /**
+     * Describes all entities matching the provided filter that have at least one client quota configuration
+     * value defined.
+     * <p>
+     * The following exceptions can be anticipated when calling {@code get()} on the future from the
+     * returned {@link DescribeClientQuotasResult}:
+     * <ul>
+     *   <li>{@link org.apache.kafka.common.errors.ClusterAuthorizationException}
+     *   If the authenticated user didn't have describe access to the cluster.</li>
+     *   <li>{@link org.apache.kafka.common.errors.InvalidRequestException}
+     *   If the request details are invalid. e.g., an invalid entity type was specified.</li>
+     *   <li>{@link org.apache.kafka.common.errors.TimeoutException}
+     *   If the request timed out before the describe could finish.</li>
+     * </ul>
+     * <p>
+     * This operation is supported by brokers with version 2.6.0 or higher.
+     *
+     * @param filter the filter to apply to match entities
+     * @param options the options to use
+     * @return the DescribeClientQuotasResult containing the result
+     */
+    DescribeClientQuotasResult describeClientQuotas(ClientQuotaFilter filter, DescribeClientQuotasOptions options);
+
+    /**
+     * Alters client quota configurations with the specified alterations.
+     * <p>
+     * This is a convenience method for {@link #alterClientQuotas(Collection, AlterClientQuotasOptions)}
+     * with default options. See the overload for more details.
+     * <p>
+     * This operation is supported by brokers with version 2.6.0 or higher.
+     *
+     * @param entries the alterations to perform
+     * @return the AlterClientQuotasResult containing the result
+     */
+    default AlterClientQuotasResult alterClientQuotas(Collection<ClientQuotaAlteration> entries) {
+        return alterClientQuotas(entries, new AlterClientQuotasOptions());
+    }
+
+    /**
+     * Alters client quota configurations with the specified alterations.
+     * <p>
+     * Alterations for a single entity are atomic, but across entities is not guaranteed. The resulting
+     * per-entity error code should be evaluated to resolve the success or failure of all updates.
+     * <p>
+     * The following exceptions can be anticipated when calling {@code get()} on the futures obtained from
+     * the returned {@link AlterClientQuotasResult}:
+     * <ul>
+     *   <li>{@link org.apache.kafka.common.errors.ClusterAuthorizationException}
+     *   If the authenticated user didn't have alter access to the cluster.</li>
+     *   <li>{@link org.apache.kafka.common.errors.InvalidRequestException}
+     *   If the request details are invalid. e.g., a configuration key was specified more than once for an entity.</li>
+     *   <li>{@link org.apache.kafka.common.errors.TimeoutException}
+     *   If the request timed out before the alterations could finish. It cannot be guaranteed whether the update
+     *   succeed or not.</li>
+     * </ul>
+     * <p>
+     * This operation is supported by brokers with version 2.6.0 or higher.
+     *
+     * @param entries the alterations to perform
+     * @return the AlterClientQuotasResult containing the result
+     */
+    AlterClientQuotasResult alterClientQuotas(Collection<ClientQuotaAlteration> entries, AlterClientQuotasOptions options);
+
     /**
      * Get the metrics kept by the adminClient
      */

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

@@ -0,0 +1,46 @@
+/*
+ * 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#alterClientQuotas(Collection, AlterClientQuotasOptions)}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class AlterClientQuotasOptions extends AbstractOptions<AlterClientQuotasOptions> {
+
+    private boolean validateOnly = false;
+
+    /**
+     * Returns whether the request should be validated without altering the configs.
+     */
+    public boolean validateOnly() {
+        return this.validateOnly;
+    }
+
+    /**
+     * Sets whether the request should be validated without altering the configs.
+     */
+    public AlterClientQuotasOptions validateOnly(boolean validateOnly) {
+        this.validateOnly = validateOnly;
+        return this;
+    }
+}

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

@@ -0,0 +1,58 @@
+/*
+ * 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.quota.ClientQuotaEntity;
+
+import java.util.Map;
+
+/**
+ * The result of the {@link Admin#alterClientQuotas(Collection, AlterClientQuotasOptions)} call.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class AlterClientQuotasResult {
+
+    private final Map<ClientQuotaEntity, KafkaFuture<Void>> futures;
+
+    /**
+     * Maps an entity to its alteration result.
+     *
+     * @param futures maps entity to its alteration result
+     */
+    public AlterClientQuotasResult(Map<ClientQuotaEntity, KafkaFuture<Void>> futures) {
+        this.futures = futures;
+    }
+
+    /**
+     * Returns a map from quota entity to a future which can be used to check the status of the operation.
+     */
+    public Map<ClientQuotaEntity, KafkaFuture<Void>> values() {
+        return futures;
+    }
+
+    /**
+     * Returns a future which succeeds only if all quota alterations succeed.
+     */
+    public KafkaFuture<Void> all() {
+        return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0]));
+    }
+}

clients/src/main/java/org/apache/kafka/clients/admin/DescribeClientQuotasOptions.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#describeClientQuotas(ClientQuotaFilter, DescribeClientQuotasOptions)}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeClientQuotasOptions extends AbstractOptions<DescribeClientQuotasOptions> {
+}

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

@@ -0,0 +1,52 @@
+/*
+ * 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.quota.ClientQuotaEntity;
+
+import java.util.Map;
+
+/**
+ * The result of the {@link Admin#describeClientQuotas(ClientQuotaFilter, DescribeClientQuotasOptions)} call.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeClientQuotasResult {
+
+    private final KafkaFuture<Map<ClientQuotaEntity, Map<String, Double>>> entities;
+
+    /**
+     * Maps an entity to its configured quota value(s). Note if no value is defined for a quota
+     * type for that entity's config, then it is not included in the resulting value map.
+     *
+     * @param entities future for the collection of entities that matched the filter
+     */
+    public DescribeClientQuotasResult(KafkaFuture<Map<ClientQuotaEntity, Map<String, Double>>> entities) {
+        this.entities = entities;
+    }
+
+    /**
+     * Returns a map from quota entity to a future which can be used to check the status of the operation.
+     */
+    public KafkaFuture<Map<ClientQuotaEntity, Map<String, Double>>> entities() {
+        return entities;
+    }
+}