kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ij...@apache.org
Subject [kafka] 01/02: MINOR: add ConfigUtils method for printing configurations (#10714)
Date Tue, 08 Jun 2021 13:48:06 GMT
This is an automated email from the ASF dual-hosted git repository.

ijuma pushed a commit to branch 2.8
in repository https://gitbox.apache.org/repos/asf/kafka.git

commit 7f64a1c8bf25724eb56366695f7c3a645a2e7dcb
Author: Colin Patrick McCabe <cmccabe@confluent.io>
AuthorDate: Wed May 19 11:03:49 2021 -0700

    MINOR: add ConfigUtils method for printing configurations (#10714)
    
    Reviewers: Luke Chen <showuon@gmail.com>, David Arthur <mumrah@gmail.com>
---
 .../org/apache/kafka/common/config/ConfigDef.java  |  6 ++++-
 .../org/apache/kafka/common/utils/ConfigUtils.java | 31 ++++++++++++++++++++++
 .../apache/kafka/common/utils/ConfigUtilsTest.java | 28 +++++++++++++++++++
 .../scala/kafka/server/DynamicBrokerConfig.scala   |  5 ++--
 4 files changed, 67 insertions(+), 3 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
index 156e08f..4fd6954 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
@@ -802,7 +802,11 @@ public class ConfigDef {
      * The config types
      */
     public enum Type {
-        BOOLEAN, STRING, INT, SHORT, LONG, DOUBLE, LIST, CLASS, PASSWORD
+        BOOLEAN, STRING, INT, SHORT, LONG, DOUBLE, LIST, CLASS, PASSWORD;
+
+        public boolean isSensitive() {
+            return this == PASSWORD;
+        }
     }
 
     /**
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ConfigUtils.java b/clients/src/main/java/org/apache/kafka/common/utils/ConfigUtils.java
index 504a1f0..0f839ff 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/ConfigUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/ConfigUtils.java
@@ -17,10 +17,15 @@
 
 package org.apache.kafka.common.utils;
 
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.ConfigKey;
+import org.apache.kafka.common.config.ConfigDef.Type;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -113,4 +118,30 @@ public class ConfigUtils {
 
         return newConfigs;
     }
+
+    public static String configMapToRedactedString(Map<String, Object> map, ConfigDef
configDef) {
+        StringBuilder bld = new StringBuilder("{");
+        List<String> keys = new ArrayList<>(map.keySet());
+        Collections.sort(keys);
+        String prefix = "";
+        for (String key : keys) {
+            bld.append(prefix).append(key).append("=");
+            ConfigKey configKey = configDef.configKeys().get(key);
+            if (configKey == null || configKey.type().isSensitive()) {
+                bld.append("(redacted)");
+            } else {
+                Object value = map.get(key);
+                if (value == null) {
+                    bld.append("null");
+                } else if (configKey.type() == Type.STRING) {
+                    bld.append("\"").append(value).append("\"");
+                } else {
+                    bld.append(value);
+                }
+            }
+            prefix = ", ";
+        }
+        bld.append("}");
+        return bld.toString();
+    }
 }
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/ConfigUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/ConfigUtilsTest.java
index b7279bb..d760330 100644
--- a/clients/src/test/java/org/apache/kafka/common/utils/ConfigUtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/ConfigUtilsTest.java
@@ -17,8 +17,12 @@
 
 package org.apache.kafka.common.utils;
 
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -140,4 +144,28 @@ public class ConfigUtilsTest {
         assertNull(newConfig.get("foo.bar.deprecated"));
         assertNull(newConfig.get("foo.bar.even.more.deprecated"));
     }
+
+    private static final ConfigDef CONFIG = new ConfigDef().
+        define("myPassword", Type.PASSWORD, Importance.HIGH, "").
+        define("myString", Type.STRING, Importance.HIGH, "").
+        define("myInt", Type.INT, Importance.HIGH, "").
+        define("myString2", Type.STRING, Importance.HIGH, "");
+
+    @Test
+    public void testConfigMapToRedactedStringForEmptyMap() {
+        assertEquals("{}", ConfigUtils.
+            configMapToRedactedString(Collections.emptyMap(), CONFIG));
+    }
+
+    @Test
+    public void testConfigMapToRedactedStringWithSecrets() {
+        Map<String, Object> testMap1 = new HashMap<>();
+        testMap1.put("myString", "whatever");
+        testMap1.put("myInt", Integer.valueOf(123));
+        testMap1.put("myPassword", "foosecret");
+        testMap1.put("myString2", null);
+        testMap1.put("myUnknown", Integer.valueOf(456));
+        assertEquals("{myInt=123, myPassword=(redacted), myString=\"whatever\", myString2=null,
myUnknown=(redacted)}",
+            ConfigUtils.configMapToRedactedString(testMap1, CONFIG));
+    }
 }
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 91d214f..2cf24c8 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -34,7 +34,7 @@ import org.apache.kafka.common.metrics.MetricsReporter
 import org.apache.kafka.common.config.types.Password
 import org.apache.kafka.common.network.{ListenerName, ListenerReconfigurable}
 import org.apache.kafka.common.security.authenticator.LoginManager
-import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.utils.{ConfigUtils, Utils}
 
 import scala.collection._
 import scala.jdk.CollectionConverters._
@@ -601,7 +601,8 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends
Logging
     }
 
     if (!validateOnly) {
-      info(s"Reconfiguring $reconfigurable, updated configs: $updatedConfigNames custom configs:
$newCustomConfigs")
+      info(s"Reconfiguring $reconfigurable, updated configs: $updatedConfigNames " +
+           s"custom configs: ${ConfigUtils.configMapToRedactedString(newCustomConfigs, KafkaConfig.configDef)}")
       reconfigurable.reconfigure(newConfigs)
     }
   }

Mime
View raw message