kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rsiva...@apache.org
Subject [kafka] branch trunk updated: MINOR: Fix AdminClient.describeConfigs() of listener configs (#4747)
Date Thu, 22 Mar 2018 20:05:50 GMT
This is an automated email from the ASF dual-hosted git repository.

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


The following commit(s) were added to refs/heads/trunk by this push:
     new 57b1c28  MINOR: Fix AdminClient.describeConfigs() of listener configs (#4747)
57b1c28 is described below

commit 57b1c28d60e05f2cf7394ef727a86df92b1bb02b
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
AuthorDate: Thu Mar 22 20:05:45 2018 +0000

    MINOR: Fix AdminClient.describeConfigs() of listener configs (#4747)
    
    Don't return config values from `describeConfigs` if the config type cannot be determined.
Obtain config types correctly for listener configs for `describeConfigs` and password encryption.
    
    Reviewers: Jason Gustafson <jason@confluent.io>
---
 .../src/main/scala/kafka/server/AdminManager.scala | 10 ++++--
 .../scala/kafka/server/DynamicBrokerConfig.scala   | 27 ++++++++------
 .../server/DynamicBrokerReconfigurationTest.scala  | 41 ++++++++++++++++------
 .../kafka/server/DynamicBrokerConfigTest.scala     | 18 ++++++++--
 4 files changed, 71 insertions(+), 25 deletions(-)

diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 9014fab..b54defc 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -473,8 +473,14 @@ class AdminManager(val config: KafkaConfig,
                                      (name: String, value: Any): DescribeConfigsResponse.ConfigEntry
= {
     val allNames = brokerSynonyms(name)
     val configEntryType = configType(name, allNames)
-    val isSensitive = configEntryType == ConfigDef.Type.PASSWORD
-    val valueAsString = if (isSensitive) null else ConfigDef.convertToString(value, configEntryType)
+    // If we can't determine the config entry type, treat it as a sensitive config to be
safe
+    val isSensitive = configEntryType == ConfigDef.Type.PASSWORD || configEntryType == null
+    val valueAsString = if (isSensitive)
+      null
+    else value match {
+      case v: String => v
+      case _ => ConfigDef.convertToString(value, configEntryType)
+    }
     val allSynonyms = configSynonyms(name, allNames, isSensitive)
         .filter(perBrokerConfig || _.source == ConfigSource.DYNAMIC_DEFAULT_BROKER_CONFIG)
     val synonyms = if (!includeSynonyms) List.empty else allSynonyms
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 92fd5d7..766907a 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -85,14 +85,17 @@ object DynamicBrokerConfig {
 
   private val PerBrokerConfigs = DynamicSecurityConfigs  ++
     DynamicListenerConfig.ReconfigurableConfigs
+  private val ListenerMechanismConfigs = Set(KafkaConfig.SaslJaasConfigProp)
 
   val ListenerConfigRegex = """listener\.name\.[^.]*\.(.*)""".r
 
-  private[server] val DynamicPasswordConfigs = {
+  private val DynamicPasswordConfigs = {
     val passwordConfigs = KafkaConfig.configKeys.filter(_._2.`type` == ConfigDef.Type.PASSWORD).keySet
     AllDynamicConfigs.intersect(passwordConfigs)
   }
 
+  def isPasswordConfig(name: String): Boolean = DynamicBrokerConfig.DynamicPasswordConfigs.exists(name.endsWith)
+
   def brokerConfigSynonyms(name: String, matchListenerOverride: Boolean): List[String] =
{
     name match {
       case KafkaConfig.LogRollTimeMillisProp | KafkaConfig.LogRollTimeHoursProp =>
@@ -103,7 +106,12 @@ object DynamicBrokerConfig {
         List(KafkaConfig.LogFlushIntervalMsProp, KafkaConfig.LogFlushSchedulerIntervalMsProp)
       case KafkaConfig.LogRetentionTimeMillisProp | KafkaConfig.LogRetentionTimeMinutesProp
| KafkaConfig.LogRetentionTimeHoursProp =>
         List(KafkaConfig.LogRetentionTimeMillisProp, KafkaConfig.LogRetentionTimeMinutesProp,
KafkaConfig.LogRetentionTimeHoursProp)
-      case ListenerConfigRegex(baseName) if matchListenerOverride => List(name, baseName)
+      case ListenerConfigRegex(baseName) if matchListenerOverride =>
+        // `ListenerMechanismConfigs` are specified as listenerPrefix.mechanism.<configName>
+        // and other listener configs are specified as listenerPrefix.<configName>
+        // Add <configName> as a synonym in both cases.
+        val mechanismConfig = ListenerMechanismConfigs.find(baseName.endsWith)
+        List(name, mechanismConfig.getOrElse(baseName))
       case _ => List(name)
     }
   }
@@ -220,15 +228,14 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends
Logging
   private[server] def toPersistentProps(configProps: Properties, perBrokerConfig: Boolean):
Properties = {
     val props = configProps.clone().asInstanceOf[Properties]
 
-    def encodePassword(configName: String): Unit = {
-      val value = props.getProperty(configName)
+    def encodePassword(configName: String, value: String): Unit = {
       if (value != null) {
         if (!perBrokerConfig)
           throw new ConfigException("Password config can be defined only at broker level")
         props.setProperty(configName, passwordEncoder.encode(new Password(value)))
       }
     }
-    DynamicPasswordConfigs.foreach(encodePassword)
+    configProps.asScala.filterKeys(isPasswordConfig).foreach { case (name, value) => encodePassword(name,
value) }
     props
   }
 
@@ -250,8 +257,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends
Logging
     if (!perBrokerConfig)
       removeInvalidProps(perBrokerConfigs(props), "Per-broker configs defined at default
cluster level will be ignored")
 
-    def decodePassword(configName: String): Unit = {
-      val value = props.getProperty(configName)
+    def decodePassword(configName: String, value: String): Unit = {
       if (value != null) {
         try {
           props.setProperty(configName, passwordEncoder.decode(value).value)
@@ -263,7 +269,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends
Logging
       }
     }
 
-    DynamicPasswordConfigs.foreach(decodePassword)
+    props.asScala.filterKeys(isPasswordConfig).foreach { case (name, value) => decodePassword(name,
value) }
     props
   }
 
@@ -273,10 +279,9 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends
Logging
   // have been removed during broker restart.
   private def maybeReEncodePasswords(persistentProps: Properties, adminZkClient: AdminZkClient):
Properties = {
     val props = persistentProps.clone().asInstanceOf[Properties]
-    if (!props.asScala.keySet.exists(DynamicPasswordConfigs.contains)) {
+    if (props.asScala.keySet.exists(isPasswordConfig)) {
       maybeCreatePasswordEncoder(kafkaConfig.passwordEncoderOldSecret).foreach { passwordDecoder
=>
-        DynamicPasswordConfigs.foreach { configName =>
-          val value = props.getProperty(configName)
+        persistentProps.asScala.filterKeys(isPasswordConfig).foreach { case (configName,
value) =>
           if (value != null) {
             val decoded = try {
               Some(passwordDecoder.decode(value).value)
diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
index 833a98e..e207de8 100644
--- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
@@ -667,10 +667,12 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness
with SaslSet
     // ZK with newly encoded values using password.encoder.secret.
     servers.foreach { server =>
       val props = adminZkClient.fetchEntityConfig(ConfigType.Broker, server.config.brokerId.toString)
+      val propsEncodedWithOldSecret = props.clone().asInstanceOf[Properties]
       val config = server.config
       val secret = config.passwordEncoderSecret.getOrElse(throw new IllegalStateException("Password
encoder secret not configured"))
       val oldSecret = config.passwordEncoderOldSecret.getOrElse(throw new IllegalStateException("Password
encoder old secret not configured"))
-      val passwordConfigs = props.asScala.filterKeys(DynamicBrokerConfig.DynamicPasswordConfigs.contains)
+      val passwordConfigs = props.asScala.filterKeys(DynamicBrokerConfig.isPasswordConfig)
+      assertTrue("Password configs not found", passwordConfigs.nonEmpty)
       val passwordDecoder = new PasswordEncoder(secret,
         config.passwordEncoderKeyFactoryAlgorithm,
         config.passwordEncoderCipherAlgorithm,
@@ -682,18 +684,19 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness
with SaslSet
         config.passwordEncoderKeyLength,
         config.passwordEncoderIterations)
       passwordConfigs.foreach { case (name, value) =>
-          val decoded = passwordDecoder.decode(value).value
-          props.put(name, passwordEncoder.encode(new Password(decoded)))
+        val decoded = passwordDecoder.decode(value).value
+        propsEncodedWithOldSecret.put(name, passwordEncoder.encode(new Password(decoded)))
       }
       val brokerId = server.config.brokerId
-      adminZkClient.changeBrokerConfig(Seq(brokerId), props)
+      adminZkClient.changeBrokerConfig(Seq(brokerId), propsEncodedWithOldSecret)
       val updatedProps = adminZkClient.fetchEntityConfig(ConfigType.Broker, brokerId.toString)
-      passwordConfigs.foreach { case (name, value) => assertNotEquals(value, updatedProps.get(name))
}
+      passwordConfigs.foreach { case (name, value) => assertNotEquals(props.get(value),
updatedProps.get(name)) }
 
       server.startup()
       TestUtils.retry(10000) {
         val newProps = adminZkClient.fetchEntityConfig(ConfigType.Broker, brokerId.toString)
-        passwordConfigs.foreach { case (name, value) => assertEquals(value, newProps.get(name))
}
+        passwordConfigs.foreach { case (name, value) =>
+          assertEquals(passwordDecoder.decode(value), passwordDecoder.decode(newProps.getProperty(name)))
}
       }
     }
 
@@ -725,7 +728,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with
SaslSet
       .map { case (name, protocol) => s"${name.value}:${protocol.name}" }
       .mkString(",") + s",$listenerName:${securityProtocol.name}"
 
-    val props =  adminZkClient.fetchEntityConfig(ConfigType.Broker, config.brokerId.toString)
+    val props = fetchBrokerConfigsFromZooKeeper(servers.head)
     props.put(KafkaConfig.ListenersProp, listeners)
     props.put(KafkaConfig.ListenerSecurityProtocolMapProp, listenerMap)
     securityProtocol match {
@@ -739,6 +742,10 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with
SaslSet
       case SecurityProtocol.PLAINTEXT => // no additional props
     }
 
+    // Add a config to verify that configs whose types are not known are not returned by
describeConfigs()
+    val unknownConfig = "some.config"
+    props.put(unknownConfig, "some.config.value")
+
     alterConfigs(adminClients.head, props, perBrokerConfig = true).all.get
 
     TestUtils.waitUntilTrue(() => servers.forall(server => server.config.listeners.size
== existingListenerCount + 1),
@@ -755,6 +762,15 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with
SaslSet
       saslMechanisms.foreach(mechanism => verifyListener(securityProtocol, Some(mechanism)))
     else
       verifyListener(securityProtocol, None)
+
+    val brokerConfigs = describeConfig(adminClients.head).entries.asScala
+    props.asScala.foreach { case (name, value) =>
+      val entry = brokerConfigs.find(_.name == name).getOrElse(throw new IllegalArgumentException(s"Config
not found $name"))
+      if (DynamicBrokerConfig.isPasswordConfig(name) || name == unknownConfig)
+        assertNull(s"Password or unknown config returned $entry", entry.value)
+      else
+        assertEquals(value, entry.value)
+    }
   }
 
   private def verifyRemoveListener(listenerName: String, securityProtocol: SecurityProtocol,
@@ -778,7 +794,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with
SaslSet
       .map { case (listenerName, protocol) => s"${listenerName.value}:${protocol.name}"
}
       .mkString(",")
 
-    val props = adminZkClient.fetchEntityConfig(ConfigType.Broker, config.brokerId.toString)
+    val props = fetchBrokerConfigsFromZooKeeper(servers.head)
     val listenerProps = props.asScala.keySet.filter(_.startsWith(new ListenerName(listenerName).configPrefix))
     listenerProps.foreach(props.remove)
     props.put(KafkaConfig.ListenersProp, listeners)
@@ -811,6 +827,11 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with
SaslSet
     verifyProduceConsume(producer, consumer, numRecords = 10, topic)
   }
 
+  private def fetchBrokerConfigsFromZooKeeper(server: KafkaServer): Properties = {
+    val props = adminZkClient.fetchEntityConfig(ConfigType.Broker, server.config.brokerId.toString)
+    server.config.dynamicConfig.fromPersistentProps(props, perBrokerConfig = true)
+  }
+
   private def bootstrapServers: String = TestUtils.bootstrapServers(servers, new ListenerName(SecureExternal))
 
   private def createProducer(trustStore: File, retries: Int,
@@ -1090,9 +1111,9 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness
with SaslSet
   private def configureDynamicKeystoreInZooKeeper(kafkaConfig: KafkaConfig, brokers: Seq[Int],
sslProperties: Properties): Unit = {
     val keystoreProps = new Properties
     addKeystoreWithListenerPrefix(sslProperties, keystoreProps, SecureExternal)
-    kafkaConfig.dynamicConfig.toPersistentProps(keystoreProps, perBrokerConfig = true)
+    val persistentProps = kafkaConfig.dynamicConfig.toPersistentProps(keystoreProps, perBrokerConfig
= true)
     zkClient.makeSurePersistentPathExists(ConfigEntityChangeNotificationZNode.path)
-    adminZkClient.changeBrokerConfig(brokers, keystoreProps)
+    adminZkClient.changeBrokerConfig(brokers, persistentProps)
   }
 
   private def waitForConfig(propName: String, propValue: String, maxWaitMs: Long = 10000):
Unit = {
diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
index bca98d2..5c88bf2 100755
--- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
@@ -142,14 +142,16 @@ class DynamicBrokerConfigTest {
   }
 
   private def verifyConfigUpdate(name: String, value: Object, perBrokerConfig: Boolean, expectFailure:
Boolean) {
-    val config = KafkaConfig(TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port
= 8181))
+    val configProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+    configProps.put(KafkaConfig.PasswordEncoderSecretProp, "broker.secret")
+    val config = KafkaConfig(configProps)
     val props = new Properties
     props.put(name, value)
     val oldValue = config.originals.get(name)
 
     def updateConfig() = {
       if (perBrokerConfig)
-        config.dynamicConfig.updateBrokerConfig(0, props)
+        config.dynamicConfig.updateBrokerConfig(0, config.dynamicConfig.toPersistentProps(props,
perBrokerConfig))
       else
         config.dynamicConfig.updateDefaultConfig(props)
     }
@@ -267,6 +269,18 @@ class DynamicBrokerConfigTest {
   }
 
   @Test
+  def testSynonyms(): Unit = {
+    assertEquals(List("listener.name.secure.ssl.keystore.type", "ssl.keystore.type"),
+      DynamicBrokerConfig.brokerConfigSynonyms("listener.name.secure.ssl.keystore.type",
matchListenerOverride = true))
+    assertEquals(List("listener.name.sasl_ssl.plain.sasl.jaas.config", "sasl.jaas.config"),
+      DynamicBrokerConfig.brokerConfigSynonyms("listener.name.sasl_ssl.plain.sasl.jaas.config",
matchListenerOverride = true))
+    assertEquals(List("some.config"),
+      DynamicBrokerConfig.brokerConfigSynonyms("some.config", matchListenerOverride = true))
+    assertEquals(List(KafkaConfig.LogRollTimeMillisProp, KafkaConfig.LogRollTimeHoursProp),
+      DynamicBrokerConfig.brokerConfigSynonyms(KafkaConfig.LogRollTimeMillisProp, matchListenerOverride
= true))
+  }
+
+  @Test
   def testDynamicConfigInitializationWithoutConfigsInZK(): Unit = {
     val zkClient = EasyMock.createMock(classOf[kafka.zk.KafkaZkClient])
     EasyMock.expect(zkClient.getEntityConfigs(EasyMock.anyString(), EasyMock.anyString())).andReturn(new
java.util.Properties()).anyTimes()

-- 
To stop receiving notification emails like this one, please contact
rsivaram@apache.org.

Mime
View raw message