diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java b/clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java index 42cc6274980e7..187a3c54d9b52 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java +++ b/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 + 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 469476d010a1d..2ae64dee6dfd2 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 @@ -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; + } } /** 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 new file mode 100644 index 0000000000000..0f839ffa962c5 --- /dev/null +++ b/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 Map translateDeprecatedConfigs(Map 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 Map translateDeprecatedConfigs(Map configs, + Map> aliasGroups) { + Set aliasSet = Stream.concat( + aliasGroups.keySet().stream(), + aliasGroups.values().stream().flatMap(Collection::stream)) + .collect(Collectors.toSet()); + + // pass through all configurations without aliases + Map 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 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 map, ConfigDef configDef) { + StringBuilder bld = new StringBuilder("{"); + List 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 new file mode 100644 index 0000000000000..b0638b472b0dd --- /dev/null +++ b/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 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 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 config = new HashMap<>(); + config.put("foo.bar", "baz"); + Map 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 config = new HashMap<>(); + config.put("foo.bar.deprecated", null); + config.put("foo.bar", "baz"); + Map 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 config = new HashMap<>(); + config.put("foo.bar.deprecated", "baz"); + config.put("foo.bar", null); + Map 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 config = new HashMap<>(); + config.put("other", null); + Map newConfig = ConfigUtils.translateDeprecatedConfigs(config, new String[][]{ + {"foo.bar", "foo.bar.deprecated"} + }); + assertNotNull(newConfig); + assertNull(newConfig.get("other")); + } + + @Test + public void testDuplicateSynonyms() { + Map config = new HashMap<>(); + config.put("foo.bar", "baz"); + config.put("foo.bar.deprecated", "derp"); + Map 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 config = new HashMap<>(); + config.put("foo.bar.deprecated", "derp"); + config.put("foo.bar.even.more.deprecated", "very old configuration"); + Map 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 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)); + } +} diff --git a/core/src/main/scala/kafka/log/LogConfig.scala b/core/src/main/scala/kafka/log/LogConfig.scala index 59fe81cc1580b..834e7f8f59cf5 100755 --- a/core/src/main/scala/kafka/log/LogConfig.scala +++ b/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 { diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index 70061fd90c50f..7c9bd3ccb0c9a 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/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) diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala index 9c3eca735e18f..9a576b6dd9fd9 100644 --- a/core/src/main/scala/kafka/network/RequestChannel.scala +++ b/core/src/main/scala/kafka/network/RequestChannel.scala @@ -24,11 +24,9 @@ import java.util.concurrent._ import com.typesafe.scalalogging.Logger import com.yammer.metrics.core.Gauge import com.yammer.metrics.core.Meter -import kafka.log.LogConfig import kafka.metrics.KafkaMetricsGroup import kafka.server.KafkaConfig import kafka.utils.{Logging, NotNothing, Pool} -import org.apache.kafka.common.config.types.Password import org.apache.kafka.common.config.ConfigResource import org.apache.kafka.common.memory.MemoryPool import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData @@ -117,21 +115,11 @@ object RequestChannel extends Logging { def loggableRequest: AbstractRequest = { - def loggableValue(resourceType: ConfigResource.Type, name: String, value: String): String = { - val maybeSensitive = resourceType match { - case ConfigResource.Type.BROKER => KafkaConfig.maybeSensitive(KafkaConfig.configType(name)) - case ConfigResource.Type.TOPIC => KafkaConfig.maybeSensitive(LogConfig.configType(name)) - case ConfigResource.Type.BROKER_LOGGER => false - case _ => true - } - if (maybeSensitive) Password.HIDDEN else value - } - bodyAndSize.request match { case alterConfigs: AlterConfigsRequest => val loggableConfigs = alterConfigs.configs().asScala.map { case (resource, config) => val loggableEntries = new AlterConfigsRequest.Config(config.entries.asScala.map { entry => - new AlterConfigsRequest.ConfigEntry(entry.name, loggableValue(resource.`type`, entry.name, entry.value)) + new AlterConfigsRequest.ConfigEntry(entry.name, KafkaConfig.loggableValue(resource.`type`, entry.name, entry.value)) }.asJavaCollection) (resource, loggableEntries) }.asJava @@ -146,7 +134,7 @@ object RequestChannel extends Logging { resource.configs.asScala.foreach { config => newResource.configs.add(new AlterableConfig() .setName(config.name) - .setValue(loggableValue(ConfigResource.Type.forId(resource.resourceType), config.name, config.value)) + .setValue(KafkaConfig.loggableValue(ConfigResource.Type.forId(resource.resourceType), config.name, config.value)) .setConfigOperation(config.configOperation)) } resources.add(newResource) diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala index 72b8d9830c513..25c6625906fdd 100644 --- a/core/src/main/scala/kafka/server/AdminManager.scala +++ b/core/src/main/scala/kafka/server/AdminManager.scala @@ -431,8 +431,12 @@ class AdminManager(val config: KafkaConfig, info(message) resource -> ApiError.fromThrowable(new InvalidRequestException(message, e)) case e: Throwable => + val configProps = new Properties + config.entries.asScala.filter(_.value != null).foreach { configEntry => + configProps.setProperty(configEntry.name, configEntry.value) + } // Log client errors at a lower level than unexpected exceptions - val message = s"Error processing alter configs request for resource $resource, config $config" + val message = s"Error processing alter configs request for resource $resource, config ${toLoggableProps(resource, configProps).mkString(",")}" if (e.isInstanceOf[ApiException]) info(message, e) else @@ -448,7 +452,7 @@ class AdminManager(val config: KafkaConfig, adminZkClient.validateTopicConfig(topic, configProps) validateConfigPolicy(resource, configEntriesMap) if (!validateOnly) { - info(s"Updating topic $topic with new configuration $config") + info(s"Updating topic $topic with new configuration : ${toLoggableProps(resource, configProps).mkString(",")}") adminZkClient.changeTopicConfig(topic, configProps) } @@ -464,6 +468,12 @@ class AdminManager(val config: KafkaConfig, if (!validateOnly) { if (perBrokerConfig) this.config.dynamicConfig.reloadUpdatedFilesWithoutConfigChange(configProps) + + if (perBrokerConfig) + info(s"Updating broker ${brokerId.get} with new configuration : ${toLoggableProps(resource, configProps).mkString(",")}") + else + info(s"Updating brokers with new configuration : ${toLoggableProps(resource, configProps).mkString(",")}") + adminZkClient.changeBrokerConfig(brokerId, this.config.dynamicConfig.toPersistentProps(configProps, perBrokerConfig)) } @@ -471,13 +481,23 @@ class AdminManager(val config: KafkaConfig, resource -> ApiError.NONE } + private def toLoggableProps(resource: ConfigResource, configProps: Properties): Map[String, String] = { + configProps.asScala.map { + case (key, value) => (key, KafkaConfig.loggableValue(resource.`type`, key, value)) + } + } + private def alterLogLevelConfigs(alterConfigOps: List[AlterConfigOp]): Unit = { alterConfigOps.foreach { alterConfigOp => val loggerName = alterConfigOp.configEntry().name() val logLevel = alterConfigOp.configEntry().value() alterConfigOp.opType() match { - case OpType.SET => Log4jController.logLevel(loggerName, logLevel) - case OpType.DELETE => Log4jController.unsetLogLevel(loggerName) + case OpType.SET => + info(s"Updating the log level of $loggerName to $logLevel") + Log4jController.logLevel(loggerName, logLevel) + case OpType.DELETE => + info(s"Unset the log level of $loggerName") + Log4jController.unsetLogLevel(loggerName) case _ => throw new IllegalArgumentException( s"Log level cannot be changed for OpType: ${alterConfigOp.opType()}") } diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala index 2f47c21891509..59e03e9ada466 100755 --- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala +++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala @@ -33,7 +33,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.collection.JavaConverters._ @@ -291,7 +291,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging dynamicBrokerConfigs ++= props.asScala updateCurrentConfig() } catch { - case e: Exception => error(s"Per-broker configs of $brokerId could not be applied: $persistentProps", e) + case e: Exception => error(s"Per-broker configs of $brokerId could not be applied: ${persistentProps.keys()}", e) } } @@ -302,7 +302,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging dynamicDefaultConfigs ++= props.asScala updateCurrentConfig() } catch { - case e: Exception => error(s"Cluster default configs could not be applied: $persistentProps", e) + case e: Exception => error(s"Cluster default configs could not be applied: ${persistentProps.keys()}", e) } } @@ -459,7 +459,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging } invalidProps.foreach(props.remove) val configSource = if (perBrokerConfig) "broker" else "default cluster" - error(s"Dynamic $configSource config contains invalid values: $invalidProps, these configs will be ignored", e) + error(s"Dynamic $configSource config contains invalid values: ${invalidProps.keys}, these configs will be ignored", e) } } @@ -545,7 +545,8 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging } catch { case e: Exception => if (!validateOnly) - error(s"Failed to update broker configuration with configs : ${newConfig.originalsFromThisConfig}", e) + error(s"Failed to update broker configuration with configs : " + + s"${ConfigUtils.configMapToRedactedString(newConfig.originalsFromThisConfig, KafkaConfig.configDef)}", e) throw new ConfigException("Invalid dynamic configuration", e) } } @@ -591,7 +592,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) } } diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index fe0f8dfcb9b49..3fb86e6594be6 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -24,17 +24,17 @@ import kafka.api.{ApiVersion, ApiVersionValidator, KAFKA_0_10_0_IV1, KAFKA_2_1_I import kafka.cluster.EndPoint import kafka.coordinator.group.OffsetConfig import kafka.coordinator.transaction.{TransactionLog, TransactionStateManager} +import kafka.log.LogConfig import kafka.message.{BrokerCompressionCodec, CompressionCodec, ZStdCompressionCodec} import kafka.security.authorizer.AuthorizerUtils import kafka.utils.CoreUtils import kafka.utils.Implicits._ import org.apache.kafka.clients.CommonClientConfigs import org.apache.kafka.common.Reconfigurable -import org.apache.kafka.common.config.SecurityConfig +import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource, SaslConfigs, SecurityConfig, SslClientAuth, SslConfigs, TopicConfig} import org.apache.kafka.common.config.ConfigDef.{ConfigKey, ValidList} import org.apache.kafka.common.config.internals.BrokerSecurityConfigs import org.apache.kafka.common.config.types.Password -import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, SaslConfigs, SslClientAuth, SslConfigs, TopicConfig} import org.apache.kafka.common.metrics.Sensor import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.record.{LegacyRecord, Records, TimestampType} @@ -948,7 +948,7 @@ object KafkaConfig { val PasswordEncoderKeyLengthDoc = "The key length used for encoding dynamically configured passwords." val PasswordEncoderIterationsDoc = "The iteration count used for encoding dynamically configured passwords." - private val configDef = { + private[server] val configDef = { import ConfigDef.Importance._ import ConfigDef.Range._ import ConfigDef.Type._ @@ -1261,6 +1261,16 @@ object KafkaConfig { // If we can't determine the config entry type, treat it as a sensitive config to be safe configType.isEmpty || configType.contains(ConfigDef.Type.PASSWORD) } + + def loggableValue(resourceType: ConfigResource.Type, name: String, value: String): String = { + val maybeSensitive = resourceType match { + case ConfigResource.Type.BROKER => KafkaConfig.maybeSensitive(KafkaConfig.configType(name)) + case ConfigResource.Type.TOPIC => KafkaConfig.maybeSensitive(LogConfig.configType(name)) + case ConfigResource.Type.BROKER_LOGGER => false + case _ => true + } + if (maybeSensitive) Password.HIDDEN else value + } } class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigOverride: Option[DynamicBrokerConfig]) diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index 809b5df1de819..1ed3d69edb69b 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -163,6 +163,21 @@ class LogConfigTest { assertNull(nullServerDefault) } + @Test + def testOverriddenConfigsAsLoggableString(): Unit = { + val kafkaProps = TestUtils.createBrokerConfig(nodeId = 0, zkConnect = "") + kafkaProps.put("unknown.broker.password.config", "aaaaa") + kafkaProps.put(KafkaConfig.SslKeyPasswordProp, "somekeypassword") + val kafkaConfig = KafkaConfig.fromProps(kafkaProps) + val topicOverrides = new Properties + topicOverrides.setProperty(LogConfig.MinInSyncReplicasProp, "2") + topicOverrides.setProperty(KafkaConfig.SslTruststorePasswordProp, "sometrustpasswrd") + topicOverrides.setProperty("unknown.topic.password.config", "bbbb") + val logConfig = LogConfig.fromProps(KafkaServer.copyKafkaConfigToLog(kafkaConfig), topicOverrides) + assertEquals("{min.insync.replicas=2, ssl.truststore.password=(redacted), unknown.topic.password.config=(redacted)}", + logConfig.overriddenConfigsAsLoggableString) + } + private def isValid(configValue: String): Boolean = { try { ThrottledReplicaListValidator.ensureValidString("", configValue) diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml index 6ab510bb618fc..77dd83447fc09 100644 --- a/gradle/spotbugs-exclude.xml +++ b/gradle/spotbugs-exclude.xml @@ -148,6 +148,13 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read + + + + + + +