Fix a series of secret logging leakage issues

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

@@ -171,11 +171,15 @@ public int hashCode() {
         return result;
     }
 
+    /**
+     * Override toString to redact sensitive value.
+     * WARNING, user should be responsible to set correct "senstive" field for each config entry.
+     */
     @Override
     public String toString() {
         return "ConfigEntry(" +
                 "name=" + name +
-                ", value=" + value +
+                ", value=" + (isSensitive ? "Redacted" : value) +
                 ", source=" + source +
                 ", isSensitive=" + isSensitive +
                 ", isReadOnly=" + isReadOnly +

clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java

@@ -786,7 +786,11 @@ else if (value instanceof Class)
      * 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;
+        }
     }
 
     /**

clients/src/main/java/org/apache/kafka/common/utils/ConfigUtils.java

@@ -0,0 +1,147 @@
+/*
+ * 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.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;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ConfigUtils {
+
+    private static final Logger log = LoggerFactory.getLogger(ConfigUtils.class);
+
+    /**
+     * Translates deprecated configurations into their non-deprecated equivalents
+     *
+     * This is a convenience method for {@link ConfigUtils#translateDeprecatedConfigs(Map, Map)}
+     * until we can use Java 9+ {@code Map.of(..)} and {@code Set.of(...)}
+     *
+     * @param configs the input configuration
+     * @param aliasGroups An array of arrays of synonyms.  Each synonym array begins with the non-deprecated synonym
+     *                    For example, new String[][] { { a, b }, { c, d, e} }
+     *                    would declare b as a deprecated synonym for a,
+     *                    and d and e as deprecated synonyms for c.
+     *                    The ordering of synonyms determines the order of precedence
+     *                    (e.g. the first synonym takes precedence over the second one)
+     * @return a new configuration map with deprecated  keys translated to their non-deprecated equivalents
+     */
+    public static <T> Map<String, T> translateDeprecatedConfigs(Map<String, T> configs, String[][] aliasGroups) {
+        return translateDeprecatedConfigs(configs, Stream.of(aliasGroups)
+            .collect(Collectors.toMap(x -> x[0], x -> Stream.of(x).skip(1).collect(Collectors.toList()))));
+    }
+
+    /**
+     * Translates deprecated configurations into their non-deprecated equivalents
+     *
+     * @param configs the input configuration
+     * @param aliasGroups A map of config to synonyms.  Each key is the non-deprecated synonym
+     *                    For example, Map.of(a , Set.of(b), c, Set.of(d, e))
+     *                    would declare b as a deprecated synonym for a,
+     *                    and d and e as deprecated synonyms for c.
+     *                    The ordering of synonyms determines the order of precedence
+     *                    (e.g. the first synonym takes precedence over the second one)
+     * @return a new configuration map with deprecated  keys translated to their non-deprecated equivalents
+     */
+    public static <T> Map<String, T> translateDeprecatedConfigs(Map<String, T> configs,
+                                                                Map<String, List<String>> aliasGroups) {
+        Set<String> aliasSet = Stream.concat(
+            aliasGroups.keySet().stream(),
+            aliasGroups.values().stream().flatMap(Collection::stream))
+            .collect(Collectors.toSet());
+
+        // pass through all configurations without aliases
+        Map<String, T> newConfigs = configs.entrySet().stream()
+            .filter(e -> !aliasSet.contains(e.getKey()))
+            // filter out null values
+            .filter(e -> Objects.nonNull(e.getValue()))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+        aliasGroups.forEach((target, aliases) -> {
+            List<String> deprecated = aliases.stream()
+                .filter(configs::containsKey)
+                .collect(Collectors.toList());
+
+            if (deprecated.isEmpty()) {
+                // No deprecated key(s) found.
+                if (configs.containsKey(target)) {
+                    newConfigs.put(target, configs.get(target));
+                }
+                return;
+            }
+
+            String aliasString = String.join(", ", deprecated);
+
+            if (configs.containsKey(target)) {
+                // Ignore the deprecated key(s) because the actual key was set.
+                log.error(target + " was configured, as well as the deprecated alias(es) " +
+                          aliasString + ".  Using the value of " + target);
+                newConfigs.put(target, configs.get(target));
+            } else if (deprecated.size() > 1) {
+                log.error("The configuration keys " + aliasString + " are deprecated and may be " +
+                          "removed in the future.  Additionally, this configuration is ambigous because " +
+                          "these configuration keys are all aliases for " + target + ".  Please update " +
+                          "your configuration to have only " + target + " set.");
+                newConfigs.put(target, configs.get(deprecated.get(0)));
+            } else {
+                log.warn("Configuration key " + deprecated.get(0) + " is deprecated and may be removed " +
+                         "in the future.  Please update your configuration to use " + target + " instead.");
+                newConfigs.put(target, configs.get(deprecated.get(0)));
+            }
+        });
+
+        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();
+    }
+}

clients/src/test/java/org/apache/kafka/common/utils/ConfigUtilsTest.java

@@ -0,0 +1,171 @@
+/*
+ * 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.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.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+public class ConfigUtilsTest {
+
+    @Test
+    public void testTranslateDeprecated() {
+        Map<String, Object> config = new HashMap<>();
+        config.put("foo.bar", "baz");
+        config.put("foo.bar.deprecated", "quux");
+        config.put("chicken", "1");
+        config.put("rooster", "2");
+        config.put("hen", "3");
+        config.put("heifer", "moo");
+        config.put("blah", "blah");
+        config.put("unexpected.non.string.object", 42);
+        Map<String, Object> newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{
+            {"foo.bar", "foo.bar.deprecated"},
+            {"chicken", "rooster", "hen"},
+            {"cow", "beef", "heifer", "steer"}
+        });
+        assertEquals("baz", newConfig.get("foo.bar"));
+        assertEquals(null, newConfig.get("foobar.deprecated"));
+        assertEquals("1", newConfig.get("chicken"));
+        assertEquals(null, newConfig.get("rooster"));
+        assertEquals(null, newConfig.get("hen"));
+        assertEquals("moo", newConfig.get("cow"));
+        assertEquals(null, newConfig.get("beef"));
+        assertEquals(null, newConfig.get("heifer"));
+        assertEquals(null, newConfig.get("steer"));
+        assertEquals(null, config.get("cow"));
+        assertEquals("blah", config.get("blah"));
+        assertEquals("blah", newConfig.get("blah"));
+        assertEquals(42, newConfig.get("unexpected.non.string.object"));
+        assertEquals(42, config.get("unexpected.non.string.object"));
+
+    }
+
+    @Test
+    public void testAllowsNewKey() {
+        Map<String, String> config = new HashMap<>();
+        config.put("foo.bar", "baz");
+        Map<String, String> newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{
+            {"foo.bar", "foo.bar.deprecated"},
+            {"chicken", "rooster", "hen"},
+            {"cow", "beef", "heifer", "steer"}
+        });
+        assertNotNull(newConfig);
+        assertEquals("baz", newConfig.get("foo.bar"));
+        assertNull(newConfig.get("foo.bar.deprecated"));
+    }
+
+    @Test
+    public void testAllowDeprecatedNulls() {
+        Map<String, String> config = new HashMap<>();
+        config.put("foo.bar.deprecated", null);
+        config.put("foo.bar", "baz");
+        Map<String, String> newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{
+            {"foo.bar", "foo.bar.deprecated"}
+        });
+        assertNotNull(newConfig);
+        assertEquals("baz", newConfig.get("foo.bar"));
+        assertNull(newConfig.get("foo.bar.deprecated"));
+    }
+
+    @Test
+    public void testAllowNullOverride() {
+        Map<String, String> config = new HashMap<>();
+        config.put("foo.bar.deprecated", "baz");
+        config.put("foo.bar", null);
+        Map<String, String> newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{
+            {"foo.bar", "foo.bar.deprecated"}
+        });
+        assertNotNull(newConfig);
+        assertNull(newConfig.get("foo.bar"));
+        assertNull(newConfig.get("foo.bar.deprecated"));
+    }
+
+    @Test
+    public void testNullMapEntriesWithoutAliasesDoNotThrowNPE() {
+        Map<String, String> config = new HashMap<>();
+        config.put("other", null);
+        Map<String, String> newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{
+            {"foo.bar", "foo.bar.deprecated"}
+        });
+        assertNotNull(newConfig);
+        assertNull(newConfig.get("other"));
+    }
+
+    @Test
+    public void testDuplicateSynonyms() {
+        Map<String, String> config = new HashMap<>();
+        config.put("foo.bar", "baz");
+        config.put("foo.bar.deprecated", "derp");
+        Map<String, String> newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{
+            {"foo.bar", "foo.bar.deprecated"},
+            {"chicken", "foo.bar.deprecated"}
+        });
+        assertNotNull(newConfig);
+        assertEquals("baz", newConfig.get("foo.bar"));
+        assertEquals("derp", newConfig.get("chicken"));
+        assertNull(newConfig.get("foo.bar.deprecated"));
+    }
+
+    @Test
+    public void testMultipleDeprecations() {
+        Map<String, String> config = new HashMap<>();
+        config.put("foo.bar.deprecated", "derp");
+        config.put("foo.bar.even.more.deprecated", "very old configuration");
+        Map<String, String> newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{
+            {"foo.bar", "foo.bar.deprecated", "foo.bar.even.more.deprecated"}
+        });
+        assertNotNull(newConfig);
+        assertEquals("derp", newConfig.get("foo.bar"));
+        assertNull(newConfig.get("foo.bar.deprecated"));
+        assertNull(newConfig.get("foo.bar.even.more.deprecated"));
+    }
+
+    private static final ConfigDef CONFIG = new ConfigDef().
+            define("foo", Type.PASSWORD, Importance.HIGH, "").
+            define("bar", Type.STRING, Importance.HIGH, "").
+            define("quux", Type.INT, Importance.HIGH, "").
+            define("blah", 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("bar", "whatever");
+        testMap1.put("quux", Integer.valueOf(123));
+        testMap1.put("foo", "foosecret");
+        testMap1.put("blah", null);
+        testMap1.put("quuux", Integer.valueOf(456));
+        assertEquals("{bar=\"whatever\", blah=null, foo=(redacted), quuux=(redacted), quux=123}",
+                ConfigUtils.configMapToRedactedString(testMap1, CONFIG));
+    }
+}

core/src/main/scala/kafka/log/LogConfig.scala

@@ -21,13 +21,14 @@ import java.util.{Collections, Locale, Properties}
 
 import scala.collection.JavaConverters._
 import kafka.api.{ApiVersion, ApiVersionValidator}
+import kafka.log.LogConfig.configDef
 import kafka.message.BrokerCompressionCodec
 import kafka.server.{KafkaConfig, ThrottledReplicaListValidator}
 import kafka.utils.Implicits._
 import org.apache.kafka.common.errors.InvalidConfigurationException
 import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, TopicConfig}
 import org.apache.kafka.common.record.{LegacyRecord, TimestampType}
-import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.utils.{ConfigUtils, Utils}
 
 import scala.collection.{Map, mutable}
 import org.apache.kafka.common.config.ConfigDef.{ConfigKey, ValidList, Validator}
@@ -108,6 +109,13 @@ case class LogConfig(props: java.util.Map[_, _], overriddenConfigs: Set[String]
     if (compact && maxCompactionLagMs > 0) math.min(maxCompactionLagMs, segmentMs)
     else segmentMs
   }
+
+  def overriddenConfigsAsLoggableString: String = {
+    val overriddenTopicProps = props.asScala.collect {
+      case (k: String, v) if overriddenConfigs.contains(k) => (k, v.asInstanceOf[AnyRef])
+    }
+    ConfigUtils.configMapToRedactedString(overriddenTopicProps.asJava, configDef)
+  }
 }
 
 object LogConfig {

core/src/main/scala/kafka/log/LogManager.scala

@@ -773,7 +773,7 @@ class LogManager(logDirs: Seq[File],
         else
           currentLogs.put(topicPartition, log)
 
-        info(s"Created log for partition $topicPartition in $logDir with properties " + s"{${config.originals.asScala.mkString(", ")}}.")
+        info(s"Created log for partition $topicPartition in $logDir with properties ${config.overriddenConfigsAsLoggableString}")
         // Remove the preferred log dir since it has already been satisfied
         preferredLogDirs.remove(topicPartition)