kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rsiva...@apache.org
Subject kafka git commit: KAFKA-5735; KIP-190: Handle client-ids consistently
Date Thu, 21 Sep 2017 20:17:17 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk afaaea809 -> c8c6ab632


KAFKA-5735; KIP-190: Handle client-ids consistently

Developed with edoardocomar

Author: Mickael Maison <mickael.maison@gmail.com>

Reviewers: Edoardo Comar <ecomar@uk.ibm.com>, Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #3906 from mimaison/KAFKA-5735


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/c8c6ab63
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/c8c6ab63
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/c8c6ab63

Branch: refs/heads/trunk
Commit: c8c6ab63248639b167350642efdfc4341fa3ce37
Parents: afaaea8
Author: Mickael Maison <mickael.maison@gmail.com>
Authored: Thu Sep 21 21:16:57 2017 +0100
Committer: Rajini Sivaram <rajinisivaram@googlemail.com>
Committed: Thu Sep 21 21:16:57 2017 +0100

----------------------------------------------------------------------
 .../kafka/clients/admin/KafkaAdminClient.java   | 16 ++--
 .../kafka/clients/consumer/KafkaConsumer.java   |  8 +-
 .../kafka/clients/producer/KafkaProducer.java   | 15 ++--
 .../apache/kafka/common/metrics/Sanitizer.java  | 61 +++++++++++++
 .../org/apache/kafka/common/utils/Utils.java    |  4 +-
 .../kafka/common/metrics/SanitizerTest.java     | 35 ++++++++
 .../src/main/scala/kafka/admin/AdminUtils.scala |  6 +-
 .../main/scala/kafka/admin/ConfigCommand.scala  |  8 +-
 .../scala/kafka/network/RequestChannel.scala    |  3 +-
 .../scala/kafka/server/ClientQuotaManager.scala | 94 ++++++++------------
 .../server/ClientRequestQuotaManager.scala      |  4 +-
 .../main/scala/kafka/server/ConfigHandler.scala | 24 ++---
 .../integration/kafka/api/BaseQuotaTest.scala   | 12 +--
 .../kafka/api/ClientIdQuotaTest.scala           |  9 +-
 .../kafka/api/UserClientIdQuotaTest.scala       | 13 +--
 .../integration/kafka/api/UserQuotaTest.scala   |  3 +-
 .../unit/kafka/admin/ConfigCommandTest.scala    | 37 ++++----
 .../kafka/server/ClientQuotaManagerTest.scala   | 23 +++--
 docs/upgrade.html                               |  1 +
 19 files changed, 235 insertions(+), 141 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/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 27c4b18..2447909 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
@@ -55,6 +55,7 @@ import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.metrics.Sanitizer;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.network.ChannelBuilder;
 import org.apache.kafka.common.network.Selector;
@@ -289,6 +290,7 @@ public class KafkaAdminClient extends AdminClient {
         NetworkClient networkClient = null;
         Time time = Time.SYSTEM;
         String clientId = generateClientId(config);
+        String sanitizedClientId = Sanitizer.sanitize(clientId);
         ChannelBuilder channelBuilder = null;
         Selector selector = null;
         ApiVersions apiVersions = new ApiVersions();
@@ -301,7 +303,7 @@ public class KafkaAdminClient extends AdminClient {
                     config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG), true);
             List<MetricsReporter> reporters = config.getConfiguredInstances(AdminClientConfig.METRIC_REPORTER_CLASSES_CONFIG,
                 MetricsReporter.class);
-            Map<String, String> metricTags = Collections.singletonMap("client-id", clientId);
+            Map<String, String> metricTags = Collections.singletonMap("client-id", sanitizedClientId);
             MetricConfig metricConfig = new MetricConfig().samples(config.getInt(AdminClientConfig.METRICS_NUM_SAMPLES_CONFIG))
                 .timeWindow(config.getLong(AdminClientConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
                 .recordLevel(Sensor.RecordingLevel.forName(config.getString(AdminClientConfig.METRICS_RECORDING_LEVEL_CONFIG)))
@@ -326,7 +328,7 @@ public class KafkaAdminClient extends AdminClient {
                 true,
                 apiVersions,
                 logContext);
-            return new KafkaAdminClient(config, clientId, time, metadata, metrics, networkClient,
+            return new KafkaAdminClient(config, clientId, sanitizedClientId, time, metadata, metrics, networkClient,
                 timeoutProcessorFactory, logContext);
         } catch (Throwable exc) {
             closeQuietly(metrics, "Metrics");
@@ -343,7 +345,7 @@ public class KafkaAdminClient extends AdminClient {
 
         try {
             metrics = new Metrics(new MetricConfig(), new LinkedList<MetricsReporter>(), time);
-            return new KafkaAdminClient(config, clientId, time, metadata, metrics, client, null,
+            return new KafkaAdminClient(config, clientId, Sanitizer.sanitize(clientId), time, metadata, metrics, client, null,
                     createLogContext(clientId));
         } catch (Throwable exc) {
             closeQuietly(metrics, "Metrics");
@@ -355,7 +357,7 @@ public class KafkaAdminClient extends AdminClient {
         return new LogContext("[AdminClient clientId=" + clientId + "] ");
     }
 
-    private KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
+    private KafkaAdminClient(AdminClientConfig config, String clientId, String sanitizedClientId, Time time, Metadata metadata,
                      Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory,
                      LogContext logContext) {
         this.defaultTimeoutMs = config.getInt(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG);
@@ -369,13 +371,13 @@ public class KafkaAdminClient extends AdminClient {
         this.metrics = metrics;
         this.client = client;
         this.runnable = new AdminClientRunnable();
-        String threadName = "kafka-admin-client-thread" + (clientId.length() > 0 ? " | " + clientId : "");
+        String threadName = "kafka-admin-client-thread | " + clientId;
         this.thread = new KafkaThread(threadName, runnable, true);
         this.timeoutProcessorFactory = (timeoutProcessorFactory == null) ?
             new TimeoutProcessorFactory() : timeoutProcessorFactory;
         this.maxRetries = config.getInt(AdminClientConfig.RETRIES_CONFIG);
         config.logUnused();
-        AppInfoParser.registerAppInfo(JMX_PREFIX, clientId);
+        AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId);
         log.debug("Kafka admin client initialized");
         thread.start();
     }
@@ -416,7 +418,7 @@ public class KafkaAdminClient extends AdminClient {
             // Wait for the thread to be joined.
             thread.join();
 
-            AppInfoParser.unregisterAppInfo(JMX_PREFIX, clientId);
+            AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId));
 
             log.debug("Kafka admin client closed.");
         } catch (InterruptedException e) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index 3ea0394..f4af39c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -42,6 +42,7 @@ import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.metrics.Sanitizer;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.network.ChannelBuilder;
 import org.apache.kafka.common.network.Selector;
@@ -645,6 +646,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
             if (clientId.isEmpty())
                 clientId = "consumer-" + CONSUMER_CLIENT_ID_SEQUENCE.getAndIncrement();
             this.clientId = clientId;
+            String sanitizedClientId = Sanitizer.sanitize(this.clientId);
             String groupId = config.getString(ConsumerConfig.GROUP_ID_CONFIG);
 
             LogContext logContext = new LogContext("[Consumer clientId=" + clientId + ", groupId=" + groupId + "] ");
@@ -658,7 +660,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
                 throw new ConfigException(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG + " should be greater than " + ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG + " and " + ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG);
             this.time = Time.SYSTEM;
 
-            Map<String, String> metricsTags = Collections.singletonMap("client-id", clientId);
+            Map<String, String> metricsTags = Collections.singletonMap("client-id", sanitizedClientId);
             MetricConfig metricConfig = new MetricConfig().samples(config.getInt(ConsumerConfig.METRICS_NUM_SAMPLES_CONFIG))
                     .timeWindow(config.getLong(ConsumerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
                     .recordLevel(Sensor.RecordingLevel.forName(config.getString(ConsumerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
@@ -769,7 +771,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
                     isolationLevel);
 
             config.logUnused();
-            AppInfoParser.registerAppInfo(JMX_PREFIX, clientId);
+            AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId);
 
             log.debug("Kafka consumer initialized");
         } catch (Throwable t) {
@@ -1724,7 +1726,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
         ClientUtils.closeQuietly(client, "consumer network client", firstException);
         ClientUtils.closeQuietly(keyDeserializer, "consumer key deserializer", firstException);
         ClientUtils.closeQuietly(valueDeserializer, "consumer value deserializer", firstException);
-        AppInfoParser.unregisterAppInfo(JMX_PREFIX, clientId);
+        AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId));
         log.debug("Kafka consumer has been closed");
         Throwable exception = firstException.get();
         if (exception != null && !swallowException) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 66760e2..8a1c2b7 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -51,6 +51,7 @@ import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.metrics.Sanitizer;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.network.ChannelBuilder;
 import org.apache.kafka.common.network.Selector;
@@ -233,7 +234,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
     private static final String JMX_PREFIX = "kafka.producer";
     public static final String NETWORK_THREAD_PREFIX = "kafka-producer-network-thread";
 
-    private String clientId;
+    private final String clientId;
     // Visible for testing
     final Metrics metrics;
     private final Partitioner partitioner;
@@ -312,9 +313,11 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
             Map<String, Object> userProvidedConfigs = config.originals();
             this.producerConfig = config;
             this.time = Time.SYSTEM;
-            clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG);
+            String clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG);
             if (clientId.length() <= 0)
                 clientId = "producer-" + PRODUCER_CLIENT_ID_SEQUENCE.getAndIncrement();
+            this.clientId = clientId;
+            String sanitizedClientId = Sanitizer.sanitize(clientId);
 
             String transactionalId = userProvidedConfigs.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG) ?
                     (String) userProvidedConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG) : null;
@@ -326,7 +329,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
             log = logContext.logger(KafkaProducer.class);
             log.trace("Starting the Kafka producer");
 
-            Map<String, String> metricTags = Collections.singletonMap("client-id", clientId);
+            Map<String, String> metricTags = Collections.singletonMap("client-id", sanitizedClientId);
             MetricConfig metricConfig = new MetricConfig().samples(config.getInt(ProducerConfig.METRICS_NUM_SAMPLES_CONFIG))
                     .timeWindow(config.getLong(ProducerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
                     .recordLevel(Sensor.RecordingLevel.forName(config.getString(ProducerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
@@ -420,12 +423,12 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
                     config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
                     this.transactionManager,
                     apiVersions);
-            String ioThreadName = NETWORK_THREAD_PREFIX + (clientId.length() > 0 ? " | " + clientId : "");
+            String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
             this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
             this.ioThread.start();
             this.errors = this.metrics.sensor("errors");
             config.logUnused();
-            AppInfoParser.registerAppInfo(JMX_PREFIX, clientId);
+            AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId);
             log.debug("Kafka producer started");
         } catch (Throwable t) {
             // call close methods if internal objects are already constructed this is to prevent resource leak. see KAFKA-2121
@@ -1073,7 +1076,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
         ClientUtils.closeQuietly(keySerializer, "producer keySerializer", firstException);
         ClientUtils.closeQuietly(valueSerializer, "producer valueSerializer", firstException);
         ClientUtils.closeQuietly(partitioner, "producer partitioner", firstException);
-        AppInfoParser.unregisterAppInfo(JMX_PREFIX, clientId);
+        AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId));
         log.debug("Kafka producer has been closed");
         if (firstException.get() != null && !swallowException)
             throw new KafkaException("Failed to close kafka producer", firstException.get());

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/clients/src/main/java/org/apache/kafka/common/metrics/Sanitizer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Sanitizer.java b/clients/src/main/java/org/apache/kafka/common/metrics/Sanitizer.java
new file mode 100644
index 0000000..b98a426
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Sanitizer.java
@@ -0,0 +1,61 @@
+/*
+ * 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.metrics;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.kafka.common.KafkaException;
+
+/**
+ * Utility class for sanitizing/desanitizing user principal and client-ids
+ * to a safe value for use in MetricName and as Zookeeper node name
+ */
+public class Sanitizer {
+
+    public static String sanitize(String name) {
+        String encoded = "";
+        try {
+            encoded = URLEncoder.encode(name, StandardCharsets.UTF_8.name());
+            StringBuilder builder = new StringBuilder();
+            for (int i = 0; i < encoded.length(); i++) {
+                char c = encoded.charAt(i);
+                if (c == '*') {         // Metric ObjectName treats * as pattern
+                    builder.append("%2A");
+                } else if (c == '+') {  // Space URL-encoded as +, replace with percent encoding
+                    builder.append("%20");
+                } else {
+                    builder.append(c);
+                }
+            }
+            return builder.toString();
+        } catch (UnsupportedEncodingException e) {
+            throw new KafkaException(e);
+        }
+    }
+
+    public static String desanitize(String name) {
+        try {
+            return URLDecoder.decode(name, StandardCharsets.UTF_8.name());
+        } catch (UnsupportedEncodingException e) {
+            throw new KafkaException(e);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index 1137045..82b12c3 100755
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -322,12 +322,12 @@ public class Utils {
         Class<?>[] argTypes = new Class<?>[params.length / 2];
         Object[] args = new Object[params.length / 2];
         try {
-            Class c = Class.forName(className, true, Utils.getContextOrKafkaClassLoader());
+            Class<?> c = Class.forName(className, true, Utils.getContextOrKafkaClassLoader());
             for (int i = 0; i < params.length / 2; i++) {
                 argTypes[i] = (Class<?>) params[2 * i];
                 args[i] = params[(2 * i) + 1];
             }
-            Constructor<T> constructor = c.getConstructor(argTypes);
+            Constructor<T> constructor = (Constructor<T>) c.getConstructor(argTypes);
             return constructor.newInstance(args);
         } catch (NoSuchMethodException e) {
             throw new ClassNotFoundException(String.format("Failed to find " +

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/clients/src/test/java/org/apache/kafka/common/metrics/SanitizerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/metrics/SanitizerTest.java b/clients/src/test/java/org/apache/kafka/common/metrics/SanitizerTest.java
new file mode 100644
index 0000000..d66bda1
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/metrics/SanitizerTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.metrics;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.UnsupportedEncodingException;
+
+import org.junit.Test;
+
+public class SanitizerTest {
+
+    @Test
+    public void testSanitize() throws UnsupportedEncodingException {
+        String principal = "CN=Some characters !@#$%&*()_-+=';:,/~";
+        String sanitizedPrincipal = Sanitizer.sanitize(principal);
+        assertTrue(sanitizedPrincipal.replace('%', '_').matches("[a-zA-Z0-9\\._\\-]+"));
+        assertEquals(principal, Sanitizer.desanitize(sanitizedPrincipal));
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala
index b6c9afe..8c873f7 100644
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@ -542,14 +542,14 @@ object AdminUtils extends Logging with AdminUtilities {
    * and <user> configs are not specified.
    *
    * @param zkUtils Zookeeper utilities used to write the config to ZK
-   * @param clientId: The clientId for which configs are being changed
+   * @param sanitizedClientId: The sanitized clientId for which configs are being changed
    * @param configs: The final set of configs that will be applied to the topic. If any new configs need to be added or
    *                 existing configs need to be deleted, it should be done prior to invoking this API
    *
    */
-  def changeClientIdConfig(zkUtils: ZkUtils, clientId: String, configs: Properties) {
+  def changeClientIdConfig(zkUtils: ZkUtils, sanitizedClientId: String, configs: Properties) {
     DynamicConfig.Client.validate(configs)
-    changeEntityConfig(zkUtils, ConfigType.Client, clientId, configs)
+    changeEntityConfig(zkUtils, ConfigType.Client, sanitizedClientId, configs)
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/main/scala/kafka/admin/ConfigCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 366667b..bd193c7 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -28,6 +28,7 @@ import kafka.utils.Implicits._
 import org.apache.kafka.common.security.JaasUtils
 import org.apache.kafka.common.security.scram._
 import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.metrics.Sanitizer
 import scala.collection._
 import scala.collection.JavaConverters._
 
@@ -184,7 +185,7 @@ object ConfigCommand extends Config {
       sanitizedName match {
         case Some(ConfigEntityName.Default) => "default " + typeName
         case Some(n) =>
-          val desanitized = if (entityType == ConfigType.User) QuotaId.desanitize(n) else n
+          val desanitized = if (entityType == ConfigType.User || entityType == ConfigType.Client) Sanitizer.desanitize(n) else n
           s"$typeName '$desanitized'"
         case None => entityType
       }
@@ -265,10 +266,7 @@ object ConfigCommand extends Config {
         ConfigEntityName.Default
       else {
         entityType match {
-          case ConfigType.User => QuotaId.sanitize(name)
-          case ConfigType.Client =>
-            validateChars("Client-id", name)
-            name
+          case ConfigType.User | ConfigType.Client => Sanitizer.sanitize(name)
           case _ => throw new IllegalArgumentException("Invalid entity type " + entityType)
         }
       }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/main/scala/kafka/network/RequestChannel.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index 1128fd3..e71a06b 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -27,6 +27,7 @@ import kafka.network.RequestChannel.{ShutdownRequest, BaseRequest}
 import kafka.server.QuotaId
 import kafka.utils.{Logging, NotNothing}
 import org.apache.kafka.common.memory.MemoryPool
+import org.apache.kafka.common.metrics.Sanitizer
 import org.apache.kafka.common.network.Send
 import org.apache.kafka.common.protocol.ApiKeys
 import org.apache.kafka.common.requests._
@@ -45,7 +46,7 @@ object RequestChannel extends Logging {
   case object ShutdownRequest extends BaseRequest
 
   case class Session(principal: KafkaPrincipal, clientAddress: InetAddress) {
-    val sanitizedUser = QuotaId.sanitize(principal.getName)
+    val sanitizedUser = Sanitizer.sanitize(principal.getName)
   }
 
   class Request(val processor: Int,

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/main/scala/kafka/server/ClientQuotaManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ClientQuotaManager.scala b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
index e1d5249..c84fbcb 100644
--- a/core/src/main/scala/kafka/server/ClientQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
@@ -75,36 +75,9 @@ object QuotaTypes {
   val UserClientIdQuotaEnabled = 4
 }
 
-object QuotaId {
+case class QuotaId(sanitizedUser: Option[String], sanitizedClientId: Option[String])
 
-  /**
-   * Sanitizes user principal to a safe value for use in MetricName
-   * and as Zookeeper node name
-   */
-  def sanitize(user: String): String = {
-    val encoded = URLEncoder.encode(user, StandardCharsets.UTF_8.name)
-    val builder = new StringBuilder
-    for (i <- 0 until encoded.length) {
-      encoded.charAt(i) match {
-        case '*' => builder.append("%2A") // Metric ObjectName treats * as pattern
-        case '+' => builder.append("%20") // Space URL-encoded as +, replace with percent encoding
-        case c => builder.append(c)
-      }
-    }
-    builder.toString
-  }
-
-  /**
-   * Decodes sanitized user principal
-   */
-  def desanitize(sanitizedUser: String): String = {
-    URLDecoder.decode(sanitizedUser, StandardCharsets.UTF_8.name)
-  }
-}
-
-case class QuotaId(sanitizedUser: Option[String], clientId: Option[String])
-
-case class QuotaEntity(quotaId: QuotaId, sanitizedUser: String, clientId: String, quota: Quota)
+case class QuotaEntity(quotaId: QuotaId, sanitizedUser: String, sanitizedClientId: String, quota: Quota)
 
 /**
  * Helper class that records per-client metrics. It is also responsible for maintaining Quota usage statistics
@@ -216,7 +189,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
       case _: QuotaViolationException =>
         // Compute the delay
         val clientQuotaEntity = clientSensors.quotaEntity
-        val clientMetric = metrics.metrics().get(clientRateMetricName(clientQuotaEntity.sanitizedUser, clientQuotaEntity.clientId))
+        val clientMetric = metrics.metrics().get(clientRateMetricName(clientQuotaEntity.sanitizedUser, clientQuotaEntity.sanitizedClientId))
         throttleTimeMs = throttleTime(clientMetric, getQuotaMetricConfig(clientQuotaEntity.quota)).toInt
         clientSensors.throttleTimeSensor.record(throttleTimeMs)
         // If delayed, add the element to the delayQueue
@@ -242,17 +215,17 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
    * and the associated quota override or default quota.
    *
    */
-  private def quotaEntity(sanitizedUser: String, clientId: String) : QuotaEntity = {
+  private def quotaEntity(sanitizedUser: String, sanitizedClientId: String) : QuotaEntity = {
     quotaTypesEnabled match {
       case QuotaTypes.NoQuotas | QuotaTypes.ClientIdQuotaEnabled =>
-        val quotaId = QuotaId(None, Some(clientId))
+        val quotaId = QuotaId(None, Some(sanitizedClientId))
         var quota = overriddenQuota.get(quotaId)
         if (quota == null) {
           quota = overriddenQuota.get(ClientQuotaManagerConfig.DefaultClientIdQuotaId)
           if (quota == null)
             quota = staticConfigClientIdQuota
         }
-        QuotaEntity(quotaId, "", clientId, quota)
+        QuotaEntity(quotaId, "", sanitizedClientId, quota)
       case QuotaTypes.UserQuotaEnabled =>
         val quotaId = QuotaId(Some(sanitizedUser), None)
         var quota = overriddenQuota.get(quotaId)
@@ -263,12 +236,12 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
         }
         QuotaEntity(quotaId, sanitizedUser, "", quota)
       case QuotaTypes.UserClientIdQuotaEnabled =>
-        val quotaId = QuotaId(Some(sanitizedUser), Some(clientId))
+        val quotaId = QuotaId(Some(sanitizedUser), Some(sanitizedClientId))
         var quota = overriddenQuota.get(quotaId)
         if (quota == null) {
           quota = overriddenQuota.get(QuotaId(Some(sanitizedUser), Some(ConfigEntityName.Default)))
           if (quota == null) {
-            quota = overriddenQuota.get(QuotaId(Some(ConfigEntityName.Default), Some(clientId)))
+            quota = overriddenQuota.get(QuotaId(Some(ConfigEntityName.Default), Some(sanitizedClientId)))
             if (quota == null) {
               quota = overriddenQuota.get(ClientQuotaManagerConfig.DefaultUserClientIdQuotaId)
               if (quota == null)
@@ -276,17 +249,17 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
             }
           }
         }
-        QuotaEntity(quotaId, sanitizedUser, clientId, quota)
+        QuotaEntity(quotaId, sanitizedUser, sanitizedClientId, quota)
       case _ =>
-        quotaEntityWithMultipleQuotaLevels(sanitizedUser, clientId)
+        quotaEntityWithMultipleQuotaLevels(sanitizedUser, sanitizedClientId)
     }
   }
 
-  private def quotaEntityWithMultipleQuotaLevels(sanitizedUser: String, clientId: String) : QuotaEntity = {
-    val userClientQuotaId = QuotaId(Some(sanitizedUser), Some(clientId))
+  private def quotaEntityWithMultipleQuotaLevels(sanitizedUser: String, sanitizerClientId: String) : QuotaEntity = {
+    val userClientQuotaId = QuotaId(Some(sanitizedUser), Some(sanitizerClientId))
 
     val userQuotaId = QuotaId(Some(sanitizedUser), None)
-    val clientQuotaId = QuotaId(None, Some(clientId))
+    val clientQuotaId = QuotaId(None, Some(sanitizerClientId))
     var quotaId = userClientQuotaId
     var quotaConfigId = userClientQuotaId
     // 1) /config/users/<user>/clients/<client-id>
@@ -306,7 +279,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
         if (quota == null) {
           // 4) /config/users/<default>/clients/<client-id>
           quotaId = userClientQuotaId
-          quotaConfigId = QuotaId(Some(ConfigEntityName.Default), Some(clientId))
+          quotaConfigId = QuotaId(Some(ConfigEntityName.Default), Some(sanitizerClientId))
           quota = overriddenQuota.get(quotaConfigId)
 
           if (quota == null) {
@@ -324,7 +297,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
               if (quota == null) {
                 // 7) /config/clients/<client-id>
                 quotaId = clientQuotaId
-                quotaConfigId = QuotaId(None, Some(clientId))
+                quotaConfigId = QuotaId(None, Some(sanitizerClientId))
                 quota = overriddenQuota.get(quotaConfigId)
 
                 if (quota == null) {
@@ -346,15 +319,17 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
       }
     }
     val quotaUser = if (quotaId == clientQuotaId) "" else sanitizedUser
-    val quotaClientId = if (quotaId == userQuotaId) "" else clientId
+    val quotaClientId = if (quotaId == userQuotaId) "" else sanitizerClientId
     QuotaEntity(quotaId, quotaUser, quotaClientId, quota)
   }
 
   /**
    * Returns the quota for the client with the specified (non-encoded) user principal and client-id.
+   * 
+   * Note: this method is expensive, it is meant to be used by tests only
    */
   def quota(user: String, clientId: String) = {
-    quotaEntity(QuotaId.sanitize(user), clientId).quota
+    quotaEntity(Sanitizer.sanitize(user), Sanitizer.sanitize(clientId)).quota
   }
 
   /*
@@ -387,14 +362,15 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
    * First sensor of the tuple is the quota enforcement sensor. Second one is the throttle time sensor
    */
   def getOrCreateQuotaSensors(sanitizedUser: String, clientId: String): ClientSensors = {
-    val clientQuotaEntity = quotaEntity(sanitizedUser, clientId)
+    val sanitizedClientId = Sanitizer.sanitize(clientId)
+    val clientQuotaEntity = quotaEntity(sanitizedUser, sanitizedClientId)
     // Names of the sensors to access
     ClientSensors(
       clientQuotaEntity,
       sensorAccessor.getOrCreate(
         getQuotaSensorName(clientQuotaEntity.quotaId),
         ClientQuotaManagerConfig.InactiveSensorExpirationTimeSeconds,
-        clientRateMetricName(clientQuotaEntity.sanitizedUser, clientQuotaEntity.clientId),
+        clientRateMetricName(clientQuotaEntity.sanitizedUser, clientQuotaEntity.sanitizedClientId),
         Some(getQuotaMetricConfig(clientQuotaEntity.quota)),
         new Rate
       ),
@@ -407,9 +383,9 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
     )
   }
 
-  private def getThrottleTimeSensorName(quotaId: QuotaId): String = quotaType + "ThrottleTime-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.clientId.getOrElse("")
+  private def getThrottleTimeSensorName(quotaId: QuotaId): String = quotaType + "ThrottleTime-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.sanitizedClientId.getOrElse("")
 
-  private def getQuotaSensorName(quotaId: QuotaId): String = quotaType + "-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.clientId.getOrElse("")
+  private def getQuotaSensorName(quotaId: QuotaId): String = quotaType + "-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.sanitizedClientId.getOrElse("")
 
   protected def getQuotaMetricConfig(quota: Quota): MetricConfig = {
     new MetricConfig()
@@ -432,10 +408,10 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
    * Overrides quotas for <user>, <client-id> or <user, client-id> or the dynamic defaults
    * for any of these levels.
    * @param sanitizedUser user to override if quota applies to <user> or <user, client-id>
-   * @param clientId client to override if quota applies to <client-id> or <user, client-id>
+   * @param sanitizedClientId client to override if quota applies to <client-id> or <user, client-id>
    * @param quota custom quota to apply or None if quota override is being removed
    */
-  def updateQuota(sanitizedUser: Option[String], clientId: Option[String], quota: Option[Quota]) {
+  def updateQuota(sanitizedUser: Option[String], sanitizedClientId: Option[String], quota: Option[Quota]) {
     /*
      * Acquire the write lock to apply changes in the quota objects.
      * This method changes the quota in the overriddenQuota map and applies the update on the actual KafkaMetric object (if it exists).
@@ -445,13 +421,13 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
      */
     lock.writeLock().lock()
     try {
-      val quotaId = QuotaId(sanitizedUser, clientId)
+      val quotaId = QuotaId(sanitizedUser, sanitizedClientId)
       val userInfo = sanitizedUser match {
         case Some(ConfigEntityName.Default) => "default user "
         case Some(user) => "user " + user + " "
         case None => ""
       }
-      val clientIdInfo = clientId match {
+      val clientIdInfo = sanitizedClientId match {
         case Some(ConfigEntityName.Default) => "default client-id"
         case Some(id) => "client-id " + id
         case None => ""
@@ -460,7 +436,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
         case Some(newQuota) =>
           logger.info(s"Changing ${quotaType} quota for ${userInfo}${clientIdInfo} to $newQuota.bound}")
           overriddenQuota.put(quotaId, newQuota)
-          (sanitizedUser, clientId) match {
+          (sanitizedUser, sanitizedClientId) match {
             case (Some(_), Some(_)) => quotaTypesEnabled |= QuotaTypes.UserClientIdQuotaEnabled
             case (Some(_), None) => quotaTypesEnabled |= QuotaTypes.UserQuotaEnabled
             case (None, Some(_)) => quotaTypesEnabled |= QuotaTypes.ClientIdQuotaEnabled
@@ -471,21 +447,21 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
           overriddenQuota.remove(quotaId)
       }
 
-      val quotaMetricName = clientRateMetricName(sanitizedUser.getOrElse(""), clientId.getOrElse(""))
+      val quotaMetricName = clientRateMetricName(sanitizedUser.getOrElse(""), sanitizedClientId.getOrElse(""))
       val allMetrics = metrics.metrics()
 
       // If multiple-levels of quotas are defined or if this is a default quota update, traverse metrics
       // to find all affected values. Otherwise, update just the single matching one.
       val singleUpdate = quotaTypesEnabled match {
         case QuotaTypes.NoQuotas | QuotaTypes.ClientIdQuotaEnabled | QuotaTypes.UserQuotaEnabled | QuotaTypes.UserClientIdQuotaEnabled =>
-          !sanitizedUser.filter(_ == ConfigEntityName.Default).isDefined && !clientId.filter(_ == ConfigEntityName.Default).isDefined
+          !sanitizedUser.filter(_ == ConfigEntityName.Default).isDefined && !sanitizedClientId.filter(_ == ConfigEntityName.Default).isDefined
         case _ => false
       }
       if (singleUpdate) {
           // Change the underlying metric config if the sensor has been created
           val metric = allMetrics.get(quotaMetricName)
           if (metric != null) {
-            val metricConfigEntity = quotaEntity(sanitizedUser.getOrElse(""), clientId.getOrElse(""))
+            val metricConfigEntity = quotaEntity(sanitizedUser.getOrElse(""), sanitizedClientId.getOrElse(""))
             val newQuota = metricConfigEntity.quota
             logger.info(s"Sensor for ${userInfo}${clientIdInfo} already exists. Changing quota to ${newQuota.bound()} in MetricConfig")
             metric.config(getQuotaMetricConfig(newQuota))
@@ -509,11 +485,11 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
     }
   }
 
-  protected def clientRateMetricName(sanitizedUser: String, clientId: String): MetricName = {
+  protected def clientRateMetricName(sanitizedUser: String, sanitizedClientId: String): MetricName = {
     metrics.metricName("byte-rate", quotaType.toString,
                    "Tracking byte-rate per user/client-id",
                    "user", sanitizedUser,
-                   "client-id", clientId)
+                   "client-id", sanitizedClientId)
   }
 
   private def throttleMetricName(quotaEntity: QuotaEntity): MetricName = {
@@ -521,7 +497,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
                        quotaType.toString,
                        "Tracking average throttle-time per user/client-id",
                        "user", quotaEntity.sanitizedUser,
-                       "client-id", quotaEntity.clientId)
+                       "client-id", quotaEntity.sanitizedClientId)
   }
 
   def shutdown() = {

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala b/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala
index f454483..d2114dc 100644
--- a/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala
@@ -64,11 +64,11 @@ class ClientRequestQuotaManager(private val config: ClientQuotaManagerConfig,
     math.min(super.throttleTime(clientMetric, config), maxThrottleTimeMs)
   }
 
-  override protected def clientRateMetricName(sanitizedUser: String, clientId: String): MetricName = {
+  override protected def clientRateMetricName(sanitizedUser: String, sanitizedClientId: String): MetricName = {
     metrics.metricName("request-time", QuotaType.Request.toString,
                    "Tracking request-time per user/client-id",
                    "user", sanitizedUser,
-                   "client-id", clientId)
+                   "client-id", sanitizedClientId)
   }
 
   private def exemptMetricName: MetricName = {

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/main/scala/kafka/server/ConfigHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala b/core/src/main/scala/kafka/server/ConfigHandler.scala
index 79ffde8..6f85801 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -29,7 +29,7 @@ import kafka.utils.Implicits._
 import kafka.utils.Logging
 import org.apache.kafka.common.config.ConfigDef.Validator
 import org.apache.kafka.common.config.ConfigException
-import org.apache.kafka.common.metrics.Quota
+import org.apache.kafka.common.metrics.{Quota, Sanitizer}
 import org.apache.kafka.common.metrics.Quota._
 
 import scala.collection.JavaConverters._
@@ -117,25 +117,25 @@ class TopicConfigHandler(private val logManager: LogManager, kafkaConfig: KafkaC
  */
 class QuotaConfigHandler(private val quotaManagers: QuotaManagers) {
 
-  def updateQuotaConfig(sanitizedUser: Option[String], clientId: Option[String], config: Properties) {
+  def updateQuotaConfig(sanitizedUser: Option[String], sanitizedClientId: Option[String], config: Properties) {
     val producerQuota =
       if (config.containsKey(DynamicConfig.Client.ProducerByteRateOverrideProp))
         Some(new Quota(config.getProperty(DynamicConfig.Client.ProducerByteRateOverrideProp).toLong, true))
       else
         None
-    quotaManagers.produce.updateQuota(sanitizedUser, clientId, producerQuota)
+    quotaManagers.produce.updateQuota(sanitizedUser, sanitizedClientId, producerQuota)
     val consumerQuota =
       if (config.containsKey(DynamicConfig.Client.ConsumerByteRateOverrideProp))
         Some(new Quota(config.getProperty(DynamicConfig.Client.ConsumerByteRateOverrideProp).toLong, true))
       else
         None
-    quotaManagers.fetch.updateQuota(sanitizedUser, clientId, consumerQuota)
+    quotaManagers.fetch.updateQuota(sanitizedUser, sanitizedClientId, consumerQuota)
     val requestQuota =
       if (config.containsKey(DynamicConfig.Client.RequestPercentageOverrideProp))
         Some(new Quota(config.getProperty(DynamicConfig.Client.RequestPercentageOverrideProp).toDouble, true))
       else
         None
-    quotaManagers.request.updateQuota(sanitizedUser, clientId, requestQuota)
+    quotaManagers.request.updateQuota(sanitizedUser, sanitizedClientId, requestQuota)
   }
 }
 
@@ -145,14 +145,14 @@ class QuotaConfigHandler(private val quotaManagers: QuotaManagers) {
  */
 class ClientIdConfigHandler(private val quotaManagers: QuotaManagers) extends QuotaConfigHandler(quotaManagers) with ConfigHandler {
 
-  def processConfigChanges(clientId: String, clientConfig: Properties) {
-    updateQuotaConfig(None, Some(clientId), clientConfig)
+  def processConfigChanges(sanitizedClientId: String, clientConfig: Properties) {
+    updateQuotaConfig(None, Some(sanitizedClientId), clientConfig)
   }
 }
 
 /**
  * The UserConfigHandler will process <user> and <user, client-id> quota changes in ZK.
- * The callback provides the node name containing sanitized user principal, client-id if this is
+ * The callback provides the node name containing sanitized user principal, sanitized client-id if this is
  * a <user, client-id> update and the full properties set read from ZK.
  */
 class UserConfigHandler(private val quotaManagers: QuotaManagers, val credentialProvider: CredentialProvider) extends QuotaConfigHandler(quotaManagers) with ConfigHandler {
@@ -163,10 +163,10 @@ class UserConfigHandler(private val quotaManagers: QuotaManagers, val credential
     if (entities.length != 1 && entities.length != 3)
       throw new IllegalArgumentException("Invalid quota entity path: " + quotaEntityPath)
     val sanitizedUser = entities(0)
-    val clientId = if (entities.length == 3) Some(entities(2)) else None
-    updateQuotaConfig(Some(sanitizedUser), clientId, config)
-    if (!clientId.isDefined && sanitizedUser != ConfigEntityName.Default)
-      credentialProvider.updateCredentials(QuotaId.desanitize(sanitizedUser), config)
+    val sanitizedClientId = if (entities.length == 3) Some(entities(2)) else None
+    updateQuotaConfig(Some(sanitizedUser), sanitizedClientId, config)
+    if (!sanitizedClientId.isDefined && sanitizedUser != ConfigEntityName.Default)
+      credentialProvider.updateCredentials(Sanitizer.desanitize(sanitizedUser), config)
   }
 }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
index c69c9a4..e8967d1 100644
--- a/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
@@ -22,7 +22,7 @@ import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 import org.apache.kafka.clients.producer._
 import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
 import org.apache.kafka.common.{MetricName, TopicPartition}
-import org.apache.kafka.common.metrics.{KafkaMetric, Quota}
+import org.apache.kafka.common.metrics.{KafkaMetric, Quota, Sanitizer}
 import org.junit.Assert._
 import org.junit.{Before, Test}
 
@@ -39,8 +39,8 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
   val consumerCount = 1
 
   private val producerBufferSize = 300000
-  protected val producerClientId = "QuotasTestProducer-1"
-  protected val consumerClientId = "QuotasTestConsumer-1"
+  protected def producerClientId = "QuotasTestProducer-1"
+  protected def consumerClientId = "QuotasTestConsumer-1"
 
   this.serverConfig.setProperty(KafkaConfig.ControlledShutdownEnableProp, "false")
   this.serverConfig.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp, "2")
@@ -210,7 +210,7 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
 
   private def verifyProducerThrottleTimeMetric(producer: KafkaProducer[_, _]) {
     val tags = new HashMap[String, String]
-    tags.put("client-id", producerClientId)
+    tags.put("client-id", Sanitizer.sanitize(producerClientId))
     val avgMetric = producer.metrics.get(new MetricName("produce-throttle-time-avg", "producer-metrics", "", tags))
     val maxMetric = producer.metrics.get(new MetricName("produce-throttle-time-max", "producer-metrics", "", tags))
 
@@ -220,7 +220,7 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
 
   private def verifyConsumerThrottleTimeMetric(consumer: KafkaConsumer[_, _], maxThrottleTime: Option[Double] = None) {
     val tags = new HashMap[String, String]
-    tags.put("client-id", consumerClientId)
+    tags.put("client-id", Sanitizer.sanitize(consumerClientId))
     val avgMetric = consumer.metrics.get(new MetricName("fetch-throttle-time-avg", "consumer-fetch-manager-metrics", "", tags))
     val maxMetric = consumer.metrics.get(new MetricName("fetch-throttle-time-max", "consumer-fetch-manager-metrics", "", tags))
 
@@ -234,7 +234,7 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
                                   quotaType.toString,
                                   "Tracking throttle-time per user/client-id",
                                   "user", quotaId.sanitizedUser.getOrElse(""),
-                                  "client-id", quotaId.clientId.getOrElse(""))
+                                  "client-id", quotaId.sanitizedClientId.getOrElse(""))
   }
 
   def throttleMetric(quotaType: QuotaType, quotaId: QuotaId): KafkaMetric = {

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala b/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala
index f8594e1..f5a2cf5 100644
--- a/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala
@@ -18,14 +18,17 @@ import java.util.Properties
 
 import kafka.admin.AdminUtils
 import kafka.server.{DynamicConfig, KafkaConfig, QuotaId}
+import org.apache.kafka.common.metrics.Sanitizer
 import org.apache.kafka.common.security.auth.KafkaPrincipal
 import org.junit.Before
 
 class ClientIdQuotaTest extends BaseQuotaTest {
 
   override val userPrincipal = KafkaPrincipal.ANONYMOUS.getName
-  override val producerQuotaId = QuotaId(None, Some(producerClientId))
-  override val consumerQuotaId = QuotaId(None, Some(consumerClientId))
+  override def producerClientId = "QuotasTestProducer-!@#$%^&*()"
+  override def consumerClientId = "QuotasTestConsumer-!@#$%^&*()"
+  override val producerQuotaId = QuotaId(None, Some(Sanitizer.sanitize(producerClientId)))
+  override val consumerQuotaId = QuotaId(None, Some(Sanitizer.sanitize(consumerClientId)))
 
   @Before
   override def setUp() {
@@ -51,6 +54,6 @@ class ClientIdQuotaTest extends BaseQuotaTest {
   }
 
   private def updateQuotaOverride(clientId: String, properties: Properties) {
-    AdminUtils.changeClientIdConfig(zkUtils, clientId, properties)
+    AdminUtils.changeClientIdConfig(zkUtils, Sanitizer.sanitize(clientId), properties)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala b/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala
index 333c851..cd9437c 100644
--- a/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala
@@ -23,6 +23,7 @@ import kafka.server._
 
 import org.apache.kafka.common.protocol.SecurityProtocol
 import org.junit.Before
+import org.apache.kafka.common.metrics.Sanitizer
 
 class UserClientIdQuotaTest extends BaseQuotaTest {
 
@@ -30,8 +31,10 @@ class UserClientIdQuotaTest extends BaseQuotaTest {
   override protected lazy val trustStoreFile = Some(File.createTempFile("truststore", ".jks"))
 
   override val userPrincipal = "O=A client,CN=localhost"
-  override def producerQuotaId = QuotaId(Some(QuotaId.sanitize(userPrincipal)), Some(producerClientId))
-  override def consumerQuotaId = QuotaId(Some(QuotaId.sanitize(userPrincipal)), Some(consumerClientId))
+  override def producerClientId = "QuotasTestProducer-!@#$%^&*()"
+  override def consumerClientId = "QuotasTestConsumer-!@#$%^&*()"
+  override def producerQuotaId = QuotaId(Some(Sanitizer.sanitize(userPrincipal)), Some(Sanitizer.sanitize(producerClientId)))
+  override def consumerQuotaId = QuotaId(Some(Sanitizer.sanitize(userPrincipal)), Some(Sanitizer.sanitize(consumerClientId)))
 
   @Before
   override def setUp() {
@@ -58,11 +61,11 @@ class UserClientIdQuotaTest extends BaseQuotaTest {
 
   override def removeQuotaOverrides() {
     val emptyProps = new Properties
-    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal) + "/clients/" + producerClientId, emptyProps)
-    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal) + "/clients/" + consumerClientId, emptyProps)
+    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, Sanitizer.sanitize(userPrincipal) + "/clients/" + Sanitizer.sanitize(producerClientId), emptyProps)
+    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, Sanitizer.sanitize(userPrincipal) + "/clients/" + Sanitizer.sanitize(consumerClientId), emptyProps)
   }
 
   private def updateQuotaOverride(userPrincipal: String, clientId: String, properties: Properties) {
-    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal) + "/clients/" + clientId, properties)
+    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, Sanitizer.sanitize(userPrincipal) + "/clients/" + Sanitizer.sanitize(clientId), properties)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala b/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala
index 4ad6265..330c1e0 100644
--- a/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala
@@ -22,6 +22,7 @@ import kafka.server.{ConfigEntityName, KafkaConfig, QuotaId}
 import kafka.utils.JaasTestUtils
 import org.apache.kafka.common.protocol.SecurityProtocol
 import org.junit.{After, Before}
+import org.apache.kafka.common.metrics.Sanitizer
 
 class UserQuotaTest extends BaseQuotaTest with SaslSetup {
 
@@ -66,6 +67,6 @@ class UserQuotaTest extends BaseQuotaTest with SaslSetup {
   }
 
   private def updateQuotaOverride(properties: Properties) {
-    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal), properties)
+    AdminUtils.changeUserOrUserClientIdConfig(zkUtils, Sanitizer.sanitize(userPrincipal), properties)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index af77c67..bb17d74 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -30,6 +30,7 @@ import org.junit.Assert._
 import org.junit.Test
 import scala.collection.mutable
 import scala.collection.JavaConverters._
+import org.apache.kafka.common.metrics.Sanitizer
 
 class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
   @Test
@@ -326,9 +327,9 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
     // <user> quota
     val principal = "CN=ConfigCommandTest,O=Apache,L=<default>"
-    val sanitizedPrincipal = QuotaId.sanitize(principal)
+    val sanitizedPrincipal = Sanitizer.sanitize(principal)
     assertEquals(-1, sanitizedPrincipal.indexOf('='))
-    assertEquals(principal, QuotaId.desanitize(sanitizedPrincipal))
+    assertEquals(principal, Sanitizer.desanitize(sanitizedPrincipal))
     for (opts <- Seq(describeOpts, alterOpts)) {
       checkEntity("users", Some(principal), sanitizedPrincipal, opts)
       checkEntity("users", Some(""), ConfigEntityName.Default, opts)
@@ -362,38 +363,36 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
       assertEquals(expectedEntityName, entity.fullSanitizedName)
     }
 
-    // <default> is a valid user principal (can be handled with URL-encoding),
-    // but an invalid client-id (cannot be handled since client-ids are not encoded)
-    checkEntity("users", QuotaId.sanitize("<default>"),
+    // <default> is a valid user principal and client-id (can be handled with URL-encoding),
+    checkEntity("users", Sanitizer.sanitize("<default>"),
         "--entity-type", "users", "--entity-name", "<default>",
         "--alter", "--add-config", "a=b,c=d")
-    try {
-      checkEntity("clients", QuotaId.sanitize("<default>"),
-          "--entity-type", "clients", "--entity-name", "<default>",
-          "--alter", "--add-config", "a=b,c=d")
-      fail("Did not fail with invalid client-id")
-    } catch {
-      case _: InvalidConfigException => // expected
-    }
+    checkEntity("clients", Sanitizer.sanitize("<default>"),
+        "--entity-type", "clients", "--entity-name", "<default>",
+        "--alter", "--add-config", "a=b,c=d")
 
-    checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+
+    checkEntity("users", Sanitizer.sanitize("CN=user1") + "/clients/client1",
         "--entity-type", "users", "--entity-name", "CN=user1", "--entity-type", "clients", "--entity-name", "client1",
         "--alter", "--add-config", "a=b,c=d")
-    checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+    checkEntity("users", Sanitizer.sanitize("CN=user1") + "/clients/client1",
         "--entity-name", "CN=user1", "--entity-type", "users", "--entity-name", "client1", "--entity-type", "clients",
         "--alter", "--add-config", "a=b,c=d")
-    checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+    checkEntity("users", Sanitizer.sanitize("CN=user1") + "/clients/client1",
         "--entity-type", "clients", "--entity-name", "client1", "--entity-type", "users", "--entity-name", "CN=user1",
         "--alter", "--add-config", "a=b,c=d")
-    checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+    checkEntity("users", Sanitizer.sanitize("CN=user1") + "/clients/client1",
         "--entity-name", "client1", "--entity-type", "clients", "--entity-name", "CN=user1", "--entity-type", "users",
         "--alter", "--add-config", "a=b,c=d")
-    checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients",
+    checkEntity("users", Sanitizer.sanitize("CN=user1") + "/clients",
         "--entity-type", "clients", "--entity-name", "CN=user1", "--entity-type", "users",
         "--describe")
     checkEntity("users", "/clients",
         "--entity-type", "clients", "--entity-type", "users",
         "--describe")
+    checkEntity("users", Sanitizer.sanitize("CN=user1") + "/clients/" + Sanitizer.sanitize("client1?@%"),
+        "--entity-name", "client1?@%", "--entity-type", "clients", "--entity-name", "CN=user1", "--entity-type", "users",
+        "--alter", "--add-config", "a=b,c=d")
   }
 
   @Test
@@ -413,7 +412,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
     val clientId = "a-client"
     val principal = "CN=ConfigCommandTest.testQuotaDescribeEntities , O=Apache, L=<default>"
-    val sanitizedPrincipal = QuotaId.sanitize(principal)
+    val sanitizedPrincipal = Sanitizer.sanitize(principal)
     val userClient = sanitizedPrincipal + "/clients/" + clientId
 
     var opts = Array("--entity-type", "clients", "--entity-name", clientId)

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
index f4a55ab..7c8e5bd 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
@@ -18,7 +18,7 @@ package kafka.server
 
 import java.util.Collections
 
-import org.apache.kafka.common.metrics.{MetricConfig, Metrics, Quota}
+import org.apache.kafka.common.metrics.{MetricConfig, Metrics, Quota, Sanitizer}
 import org.apache.kafka.common.utils.MockTime
 import org.junit.Assert.{assertEquals, assertTrue}
 import org.junit.{Before, Test}
@@ -374,12 +374,21 @@ class ClientQuotaManagerTest {
   }
 
   @Test
-  def testQuotaUserSanitize() {
-    val principal = "CN=Some characters !@#$%&*()_-+=';:,/~"
-    val sanitizedPrincipal = QuotaId.sanitize(principal)
-    // Apart from % used in percent-encoding all characters of sanitized principal must be characters allowed in client-id
-    ConfigCommand.validateChars("sanitized-principal", sanitizedPrincipal.replace('%', '_'))
-    assertEquals(principal, QuotaId.desanitize(sanitizedPrincipal))
+  def testSanitizeClientId() {
+    val metrics = newMetrics
+    val clientMetrics = new ClientQuotaManager(config, metrics, QuotaType.Produce, time)
+    val clientId = "client@#$%"
+    try {
+      clientMetrics.maybeRecordAndThrottle("ANONYMOUS", clientId, 100, callback)
+      
+      val throttleTimeSensor = metrics.getSensor("ProduceThrottleTime-:" + Sanitizer.sanitize(clientId))
+      assertTrue("Throttle time sensor should exist", throttleTimeSensor != null)
+
+      val byteRateSensor = metrics.getSensor("Produce-:"  + Sanitizer.sanitize(clientId))
+      assertTrue("Byte rate sensor should exist", byteRateSensor != null)
+    } finally {
+      clientMetrics.shutdown()
+    }
   }
 
   def newMetrics: Metrics = {

http://git-wip-us.apache.org/repos/asf/kafka/blob/c8c6ab63/docs/upgrade.html
----------------------------------------------------------------------
diff --git a/docs/upgrade.html b/docs/upgrade.html
index ed0f9cf..a98bdea 100644
--- a/docs/upgrade.html
+++ b/docs/upgrade.html
@@ -65,6 +65,7 @@
         <code>OffsetFetchRequest</code>, <code>OffsetRequest</code>, <code>ProducerRequest</code>, and <code>TopicMetadataRequest</code>.
         This was only intended for use on the broker, but it is no longer in use and the implementations have not been maintained.
         A stub implementation has been retained for binary compatibility.</li>
+    <li>The Java clients and tools now accept any string as a client-id.</li>
 </ul>
 
 <h5><a id="upgrade_100_new_protocols" href="#upgrade_100_new_protocols">New Protocol Versions</a></h5>


Mime
View raw message