kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [kafka] branch trunk updated: KAFKA-6246; Dynamic update of listeners and security configs (#4488)
Date Sun, 04 Feb 2018 17:19:21 GMT
This is an automated email from the ASF dual-hosted git repository.

jgus 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 4019b21  KAFKA-6246; Dynamic update of listeners and security configs (#4488)
4019b21 is described below

commit 4019b21d602c0912d9dc0286f10701e7e724f750
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
AuthorDate: Sun Feb 4 09:19:17 2018 -0800

    KAFKA-6246; Dynamic update of listeners and security configs (#4488)
    
    Dynamic update of listeners as described in KIP-226. This includes:
      - Addition of new listeners with listener-prefixed security configs
      - Removal of existing listeners
      - Password encryption
      - sasl.jaas.config property for broker's JAAS config prefixed with listener and mechanism name
---
 checkstyle/import-control.xml                      |   1 +
 .../org/apache/kafka/common/Reconfigurable.java    |   9 +-
 .../apache/kafka/common/config/AbstractConfig.java |  16 +
 .../kafka/common/network/ChannelBuilders.java      |  18 +-
 .../apache/kafka/common/network/ListenerName.java  |   4 +
 .../kafka/common/network/SaslChannelBuilder.java   |  58 +--
 .../kafka/common/network/SslChannelBuilder.java    |   4 +-
 .../apache/kafka/common/security/JaasContext.java  |  84 ++---
 .../security/authenticator/CredentialCache.java    |   8 +-
 .../security/authenticator/LoginManager.java       |   4 +-
 .../authenticator/SaslServerAuthenticator.java     |  23 +-
 .../security/scram/ScramCredentialUtils.java       |   4 +-
 .../kafka/common/security/ssl/SslFactory.java      |   7 +-
 .../kafka/common/config/AbstractConfigTest.java    |  49 +++
 .../common/network/SaslChannelBuilderTest.java     |  20 +-
 .../kafka/common/security/JaasContextTest.java     |  25 +-
 .../authenticator/SaslAuthenticatorTest.java       |  48 ++-
 .../authenticator/SaslServerAuthenticatorTest.java |  13 +-
 .../security/authenticator/TestJaasConfig.java     |  14 +
 .../scala/kafka/controller/KafkaController.scala   |  67 +++-
 .../main/scala/kafka/network/SocketServer.scala    |  56 ++-
 .../scala/kafka/security/CredentialProvider.scala  |   6 +-
 .../scala/kafka/server/DynamicBrokerConfig.scala   | 254 ++++++++++---
 core/src/main/scala/kafka/server/KafkaApis.scala   |   2 +-
 core/src/main/scala/kafka/server/KafkaConfig.scala |  99 +++--
 core/src/main/scala/kafka/server/KafkaServer.scala |   9 +-
 .../main/scala/kafka/utils/PasswordEncoder.scala   | 175 +++++++++
 core/src/main/scala/kafka/zk/KafkaZkClient.scala   |   7 +
 .../kafka/api/AdminClientIntegrationTest.scala     |   4 +-
 .../scala/integration/kafka/api/SaslSetup.scala    |  16 +-
 .../server/DynamicBrokerReconfigurationTest.scala  | 404 +++++++++++++++++++--
 ...pleListenersWithAdditionalJaasContextTest.scala |  24 +-
 ...ltipleListenersWithDefaultJaasContextTest.scala |   9 +-
 ...ListenersWithSameSecurityProtocolBaseTest.scala |  81 +++--
 .../test/scala/unit/kafka/KafkaConfigTest.scala    |  12 +-
 .../scala/unit/kafka/admin/ConfigCommandTest.scala |   4 +-
 .../unit/kafka/network/SocketServerTest.scala      |   5 +-
 .../kafka/server/DynamicBrokerConfigTest.scala     | 155 ++++++--
 .../scala/unit/kafka/server/KafkaConfigTest.scala  |  10 +
 .../unit/kafka/utils/PasswordEncoderTest.scala     | 127 +++++++
 .../test/scala/unit/kafka/utils/TestUtils.scala    |   2 +-
 41 files changed, 1574 insertions(+), 363 deletions(-)

diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 5662552..000acc3 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -50,6 +50,7 @@
     <disallow pkg="org.apache.kafka.clients" />
     <allow pkg="org.apache.kafka.common" exact-match="true" />
     <allow pkg="org.apache.kafka.common.annotation" />
+    <allow pkg="org.apache.kafka.common.config" exact-match="true" />
     <allow pkg="org.apache.kafka.common.internals" exact-match="true" />
     <allow pkg="org.apache.kafka.test" />
 
diff --git a/clients/src/main/java/org/apache/kafka/common/Reconfigurable.java b/clients/src/main/java/org/apache/kafka/common/Reconfigurable.java
index 3339dce..8db9dc2 100644
--- a/clients/src/main/java/org/apache/kafka/common/Reconfigurable.java
+++ b/clients/src/main/java/org/apache/kafka/common/Reconfigurable.java
@@ -16,6 +16,8 @@
  */
 package org.apache.kafka.common;
 
+import org.apache.kafka.common.config.ConfigException;
+
 import java.util.Map;
 import java.util.Set;
 
@@ -33,9 +35,12 @@ public interface Reconfigurable extends Configurable {
      * Validates the provided configuration. The provided map contains
      * all configs including any reconfigurable configs that may be different
      * from the initial configuration. Reconfiguration will be not performed
-     * if this method returns false or throws any exception.
+     * if this method throws any exception.
+     * @throws ConfigException if the provided configs are not valid. The exception
+     *         message from ConfigException will be returned to the client in
+     *         the AlterConfigs response.
      */
-    boolean validateReconfiguration(Map<String, ?> configs);
+    void validateReconfiguration(Map<String, ?> configs) throws ConfigException;
 
     /**
      * Reconfigures this instance with the given key-value pairs. The provided
diff --git a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
index ce3ae43..1713d78 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
@@ -204,6 +204,16 @@ public class AbstractConfig {
      * put all the remaining keys with the prefix stripped and their parsed values in the result map.
      *
      * This is useful if one wants to allow prefixed configs to override default ones.
+     * <p>
+     * Two forms of prefixes are supported:
+     * <ul>
+     *     <li>listener.name.{listenerName}.some.prop: If the provided prefix is `listener.name.{listenerName}.`,
+     *         the key `some.prop` with the value parsed using the definition of `some.prop` is returned.</li>
+     *     <li>listener.name.{listenerName}.{mechanism}.some.prop: If the provided prefix is `listener.name.{listenerName}.`,
+     *         the key `{mechanism}.some.prop` with the value parsed using the definition of `some.prop` is returned.
+     *          This is used to provide per-mechanism configs for a broker listener (e.g sasl.jaas.config)</li>
+     * </ul>
+     * </p>
      */
     public Map<String, Object> valuesWithPrefixOverride(String prefix) {
         Map<String, Object> result = new RecordingMap<>(values(), prefix, true);
@@ -213,6 +223,12 @@ public class AbstractConfig {
                 ConfigDef.ConfigKey configKey = definition.configKeys().get(keyWithNoPrefix);
                 if (configKey != null)
                     result.put(keyWithNoPrefix, definition.parseValue(configKey, entry.getValue(), true));
+                else {
+                    String keyWithNoSecondaryPrefix = keyWithNoPrefix.substring(keyWithNoPrefix.indexOf('.') + 1);
+                    configKey = definition.configKeys().get(keyWithNoSecondaryPrefix);
+                    if (configKey != null)
+                        result.put(keyWithNoPrefix, definition.parseValue(configKey, entry.getValue(), true));
+                }
             }
         }
         return result;
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
index e6df78e..80ccb7e 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
@@ -29,6 +29,9 @@ import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
 import org.apache.kafka.common.security.token.delegation.DelegationTokenCache;
 import org.apache.kafka.common.utils.Utils;
 
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 public class ChannelBuilders {
@@ -105,9 +108,20 @@ public class ChannelBuilders {
             case SASL_SSL:
             case SASL_PLAINTEXT:
                 requireNonNullMode(mode, securityProtocol);
-                JaasContext jaasContext = JaasContext.load(contextType, listenerName, configs);
+                Map<String, JaasContext> jaasContexts;
+                if (mode == Mode.SERVER) {
+                    List<String> enabledMechanisms = (List<String>) configs.get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
+                    jaasContexts = new HashMap<>(enabledMechanisms.size());
+                    for (String mechanism : enabledMechanisms)
+                        jaasContexts.put(mechanism, JaasContext.loadServerContext(listenerName, mechanism, configs));
+                } else {
+                    // Use server context for inter-broker client connections and client context for other clients
+                    JaasContext jaasContext = contextType == JaasContext.Type.CLIENT ? JaasContext.loadClientContext(configs) :
+                            JaasContext.loadServerContext(listenerName, clientSaslMechanism, configs);
+                    jaasContexts = Collections.singletonMap(clientSaslMechanism, jaasContext);
+                }
                 channelBuilder = new SaslChannelBuilder(mode,
-                        jaasContext,
+                        jaasContexts,
                         securityProtocol,
                         listenerName,
                         isInterBrokerListener,
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java b/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java
index 9da4cca..fc0cb14 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java
@@ -71,4 +71,8 @@ public final class ListenerName {
     public String configPrefix() {
         return CONFIG_STATIC_PREFIX + "." + value.toLowerCase(Locale.ROOT) + ".";
     }
+
+    public String saslMechanismConfigPrefix(String saslMechanism) {
+        return configPrefix() + saslMechanism.toLowerCase(Locale.ROOT) + ".";
+    }
 }
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
index 1716e0e..095f826 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
@@ -41,6 +41,7 @@ import java.net.Socket;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -55,18 +56,19 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
     private final boolean isInterBrokerListener;
     private final String clientSaslMechanism;
     private final Mode mode;
-    private final JaasContext jaasContext;
+    private final Map<String, JaasContext> jaasContexts;
     private final boolean handshakeRequestEnable;
     private final CredentialCache credentialCache;
     private final DelegationTokenCache tokenCache;
+    private final Map<String, LoginManager> loginManagers;
+    private final Map<String, Subject> subjects;
 
-    private LoginManager loginManager;
     private SslFactory sslFactory;
     private Map<String, ?> configs;
     private KerberosShortNamer kerberosShortNamer;
 
     public SaslChannelBuilder(Mode mode,
-                              JaasContext jaasContext,
+                              Map<String, JaasContext> jaasContexts,
                               SecurityProtocol securityProtocol,
                               ListenerName listenerName,
                               boolean isInterBrokerListener,
@@ -75,7 +77,9 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
                               CredentialCache credentialCache,
                               DelegationTokenCache tokenCache) {
         this.mode = mode;
-        this.jaasContext = jaasContext;
+        this.jaasContexts = jaasContexts;
+        this.loginManagers = new HashMap<>(jaasContexts.size());
+        this.subjects = new HashMap<>(jaasContexts.size());
         this.securityProtocol = securityProtocol;
         this.listenerName = listenerName;
         this.isInterBrokerListener = isInterBrokerListener;
@@ -89,14 +93,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
     public void configure(Map<String, ?> configs) throws KafkaException {
         try {
             this.configs = configs;
-            boolean hasKerberos;
-            if (mode == Mode.SERVER) {
-                @SuppressWarnings("unchecked")
-                List<String> enabledMechanisms = (List<String>) this.configs.get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
-                hasKerberos = enabledMechanisms == null || enabledMechanisms.contains(SaslConfigs.GSSAPI_MECHANISM);
-            } else {
-                hasKerberos = clientSaslMechanism.equals(SaslConfigs.GSSAPI_MECHANISM);
-            }
+            boolean hasKerberos = jaasContexts.containsKey(SaslConfigs.GSSAPI_MECHANISM);
 
             if (hasKerberos) {
                 String defaultRealm;
@@ -110,8 +107,14 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
                 if (principalToLocalRules != null)
                     kerberosShortNamer = KerberosShortNamer.fromUnparsedRules(defaultRealm, principalToLocalRules);
             }
-            this.loginManager = LoginManager.acquireLoginManager(jaasContext, hasKerberos, configs);
-
+            for (Map.Entry<String, JaasContext> entry : jaasContexts.entrySet()) {
+                String mechanism = entry.getKey();
+                // With static JAAS configuration, use KerberosLogin if Kerberos is enabled. With dynamic JAAS configuration,
+                // use KerberosLogin only for the LoginContext corresponding to GSSAPI
+                LoginManager loginManager = LoginManager.acquireLoginManager(entry.getValue(), mechanism, hasKerberos, configs);
+                loginManagers.put(mechanism, loginManager);
+                subjects.put(mechanism, loginManager.subject());
+            }
             if (this.securityProtocol == SecurityProtocol.SASL_SSL) {
                 // Disable SSL client authentication as we are using SASL authentication
                 this.sslFactory = new SslFactory(mode, "none", isInterBrokerListener);
@@ -129,11 +132,9 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
     }
 
     @Override
-    public boolean validateReconfiguration(Map<String, ?> configs) {
+    public void validateReconfiguration(Map<String, ?> configs) {
         if (this.securityProtocol == SecurityProtocol.SASL_SSL)
-            return sslFactory.validateReconfiguration(configs);
-        else
-            return true;
+            sslFactory.validateReconfiguration(configs);
     }
 
     @Override
@@ -154,11 +155,13 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
             Socket socket = socketChannel.socket();
             TransportLayer transportLayer = buildTransportLayer(id, key, socketChannel);
             Authenticator authenticator;
-            if (mode == Mode.SERVER)
-                authenticator = buildServerAuthenticator(configs, id, transportLayer, loginManager.subject());
-            else
+            if (mode == Mode.SERVER) {
+                authenticator = buildServerAuthenticator(configs, id, transportLayer, subjects);
+            } else {
+                LoginManager loginManager = loginManagers.get(clientSaslMechanism);
                 authenticator = buildClientAuthenticator(configs, id, socket.getInetAddress().getHostName(),
                         loginManager.serviceName(), transportLayer, loginManager.subject());
+            }
             return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize, memoryPool != null ? memoryPool : MemoryPool.NONE);
         } catch (Exception e) {
             log.info("Failed to create channel due to ", e);
@@ -168,10 +171,9 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
 
     @Override
     public void close()  {
-        if (loginManager != null) {
+        for (LoginManager loginManager : loginManagers.values())
             loginManager.release();
-            loginManager = null;
-        }
+        loginManagers.clear();
     }
 
     private TransportLayer buildTransportLayer(String id, SelectionKey key, SocketChannel socketChannel) throws IOException {
@@ -185,8 +187,8 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
 
     // Visible to override for testing
     protected SaslServerAuthenticator buildServerAuthenticator(Map<String, ?> configs, String id,
-            TransportLayer transportLayer, Subject subject) throws IOException {
-        return new SaslServerAuthenticator(configs, id, jaasContext, subject,
+            TransportLayer transportLayer, Map<String, Subject> subjects) throws IOException {
+        return new SaslServerAuthenticator(configs, id, jaasContexts, subjects,
                 kerberosShortNamer, credentialCache, listenerName, securityProtocol, transportLayer, tokenCache);
     }
 
@@ -198,8 +200,8 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
     }
 
     // Package private for testing
-    LoginManager loginManager() {
-        return loginManager;
+    Map<String, LoginManager> loginManagers() {
+        return loginManagers;
     }
 
     private static String defaultKerberosRealm() throws ClassNotFoundException, NoSuchMethodException,
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
index e024d32..941c455 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
@@ -71,8 +71,8 @@ public class SslChannelBuilder implements ChannelBuilder, ListenerReconfigurable
     }
 
     @Override
-    public boolean validateReconfiguration(Map<String, ?> configs) {
-        return sslFactory.validateReconfiguration(configs);
+    public void validateReconfiguration(Map<String, ?> configs) {
+        sslFactory.validateReconfiguration(configs);
     }
 
     @Override
diff --git a/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java b/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java
index a13acd2..46db345 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java
@@ -41,57 +41,59 @@ public class JaasContext {
     /**
      * Returns an instance of this class.
      *
-     * For contextType SERVER, the context will contain the default Configuration and the context name will be one of:
+     * The context will contain the configuration specified by the JAAS configuration property
+     * {@link SaslConfigs#SASL_JAAS_CONFIG} with prefix `listener.name.{listenerName}.{mechanism}.`
+     * with listenerName and mechanism in lower case. The context `KafkaServer` will be returned
+     * with a single login context entry loaded from the property.
+     * <p>
+     * If the property is not defined, the context will contain the default Configuration and
+     * the context name will be one of:
+     * <ol>
+     *   <li>Lowercased listener name followed by a period and the string `KafkaServer`</li>
+     *   <li>The string `KafkaServer`</li>
+     *  </ol>
+     * If both are valid entries in the default JAAS configuration, the first option is chosen.
+     * </p>
      *
-     * 1. Lowercased listener name followed by a period and the string `KafkaServer`
-     * 2. The string `KafkaServer`
-     *
-     * If both are valid entries in the JAAS configuration, the first option is chosen.
+     * @throws IllegalArgumentException if listenerName or mechanism is not defined.
+     */
+    public static JaasContext loadServerContext(ListenerName listenerName, String mechanism, Map<String, ?> configs) {
+        if (listenerName == null)
+            throw new IllegalArgumentException("listenerName should not be null for SERVER");
+        if (mechanism == null)
+            throw new IllegalArgumentException("mechanism should not be null for SERVER");
+        String globalContextName = GLOBAL_CONTEXT_NAME_SERVER;
+        String listenerContextName = listenerName.value().toLowerCase(Locale.ROOT) + "." + GLOBAL_CONTEXT_NAME_SERVER;
+        Password jaasConfigArgs = (Password) configs.get(mechanism.toLowerCase(Locale.ROOT) + "." + SaslConfigs.SASL_JAAS_CONFIG);
+        if (jaasConfigArgs == null && configs.containsKey(SaslConfigs.SASL_JAAS_CONFIG))
+            LOG.warn("Server config {} should be prefixed with SASL mechanism name, ignoring config", SaslConfigs.SASL_JAAS_CONFIG);
+        return load(Type.SERVER, listenerContextName, globalContextName, jaasConfigArgs);
+    }
+
+    /**
+     * Returns an instance of this class.
      *
-     * For contextType CLIENT, if JAAS configuration property @link SaslConfigs#SASL_JAAS_CONFIG} is specified,
+     * If JAAS configuration property @link SaslConfigs#SASL_JAAS_CONFIG} is specified,
      * the configuration object is created by parsing the property value. Otherwise, the default Configuration
      * is returned. The context name is always `KafkaClient`.
      *
-     * @throws IllegalArgumentException if JAAS configuration property is specified for contextType SERVER, if
-     * listenerName is not defined for contextType SERVER of if listenerName is defined for contextType CLIENT.
      */
-    public static JaasContext load(JaasContext.Type contextType, ListenerName listenerName,
-                                   Map<String, ?> configs) {
-        String listenerContextName;
-        String globalContextName;
-        switch (contextType) {
-            case CLIENT:
-                if (listenerName != null)
-                    throw new IllegalArgumentException("listenerName should be null for CLIENT");
-                globalContextName = GLOBAL_CONTEXT_NAME_CLIENT;
-                listenerContextName = null;
-                break;
-            case SERVER:
-                if (listenerName == null)
-                    throw new IllegalArgumentException("listenerName should not be null for SERVER");
-                globalContextName = GLOBAL_CONTEXT_NAME_SERVER;
-                listenerContextName = listenerName.value().toLowerCase(Locale.ROOT) + "." + GLOBAL_CONTEXT_NAME_SERVER;
-                break;
-            default:
-                throw new IllegalArgumentException("Unexpected context type " + contextType);
-        }
-        return load(contextType, listenerContextName, globalContextName, configs);
+    public static JaasContext loadClientContext(Map<String, ?> configs) {
+        String globalContextName = GLOBAL_CONTEXT_NAME_CLIENT;
+        Password jaasConfigArgs = (Password) configs.get(SaslConfigs.SASL_JAAS_CONFIG);
+        return load(JaasContext.Type.CLIENT, null, globalContextName, jaasConfigArgs);
     }
 
     static JaasContext load(JaasContext.Type contextType, String listenerContextName,
-                            String globalContextName, Map<String, ?> configs) {
-        Password jaasConfigArgs = (Password) configs.get(SaslConfigs.SASL_JAAS_CONFIG);
+                            String globalContextName, Password jaasConfigArgs) {
         if (jaasConfigArgs != null) {
-            if (contextType == JaasContext.Type.SERVER)
-                throw new IllegalArgumentException("JAAS config property not supported for server");
-            else {
-                JaasConfig jaasConfig = new JaasConfig(globalContextName, jaasConfigArgs.value());
-                AppConfigurationEntry[] clientModules = jaasConfig.getAppConfigurationEntry(globalContextName);
-                int numModules = clientModules == null ? 0 : clientModules.length;
-                if (numModules != 1)
-                    throw new IllegalArgumentException("JAAS config property contains " + numModules + " login modules, should be 1 module");
-                return new JaasContext(globalContextName, contextType, jaasConfig);
-            }
+            JaasConfig jaasConfig = new JaasConfig(globalContextName, jaasConfigArgs.value());
+            AppConfigurationEntry[] contextModules = jaasConfig.getAppConfigurationEntry(globalContextName);
+            if (contextModules == null || contextModules.length == 0)
+                throw new IllegalArgumentException("JAAS config property does not contain any login modules");
+            else if (contextModules.length != 1)
+                throw new IllegalArgumentException("JAAS config property contains " + contextModules.length + " login modules, should be 1 module");
+            return new JaasContext(globalContextName, contextType, jaasConfig);
         } else
             return defaultContext(contextType, listenerContextName, globalContextName);
     }
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java
index aa39101..96e7426 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java
@@ -16,18 +16,16 @@
  */
 package org.apache.kafka.common.security.authenticator;
 
-import java.util.HashMap;
-import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class CredentialCache {
 
-    private final Map<String, Cache<? extends Object>> cacheMap = new HashMap<>();
+    private final ConcurrentHashMap<String, Cache<? extends Object>> cacheMap = new ConcurrentHashMap<>();
 
     public <C> Cache<C> createCache(String mechanism, Class<C> credentialClass) {
         Cache<C> cache = new Cache<C>(credentialClass);
-        cacheMap.put(mechanism, cache);
-        return cache;
+        Cache<C> oldCache = (Cache<C>) cacheMap.putIfAbsent(mechanism, cache);
+        return oldCache == null ? cache : oldCache;
     }
 
     @SuppressWarnings("unchecked")
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java
index a576e37..dc264c8 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java
@@ -64,7 +64,7 @@ public class LoginManager {
      * shut it down when the broker or clients are closed. It's straightforward to do the former, but it's more
      * complicated to do the latter without making the consumer API more complex.
      */
-    public static LoginManager acquireLoginManager(JaasContext jaasContext, boolean hasKerberos,
+    public static LoginManager acquireLoginManager(JaasContext jaasContext, String saslMechanism, boolean hasKerberos,
                                                    Map<String, ?> configs) throws IOException, LoginException {
         synchronized (LoginManager.class) {
             // SASL_JAAS_CONFIG is only supported by clients
@@ -73,7 +73,7 @@ public class LoginManager {
             if (jaasContext.type() == JaasContext.Type.CLIENT && jaasConfigValue != null) {
                 loginManager = DYNAMIC_INSTANCES.get(jaasConfigValue);
                 if (loginManager == null) {
-                    loginManager = new LoginManager(jaasContext, hasKerberos, configs, jaasConfigValue);
+                    loginManager = new LoginManager(jaasContext, saslMechanism.equals(SaslConfigs.GSSAPI_MECHANISM), configs, jaasConfigValue);
                     DYNAMIC_INSTANCES.put(jaasConfigValue, loginManager);
                 }
             } else {
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
index 6fcca57..ca6e9d2 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
@@ -101,8 +101,8 @@ public class SaslServerAuthenticator implements Authenticator {
     private final SecurityProtocol securityProtocol;
     private final ListenerName listenerName;
     private final String connectionId;
-    private final JaasContext jaasContext;
-    private final Subject subject;
+    private final Map<String, JaasContext> jaasContexts;
+    private final Map<String, Subject> subjects;
     private final CredentialCache credentialCache;
     private final TransportLayer transportLayer;
     private final Set<String> enabledMechanisms;
@@ -128,19 +128,17 @@ public class SaslServerAuthenticator implements Authenticator {
 
     public SaslServerAuthenticator(Map<String, ?> configs,
                                    String connectionId,
-                                   JaasContext jaasContext,
-                                   Subject subject,
+                                   Map<String, JaasContext> jaasContexts,
+                                   Map<String, Subject> subjects,
                                    KerberosShortNamer kerberosNameParser,
                                    CredentialCache credentialCache,
                                    ListenerName listenerName,
                                    SecurityProtocol securityProtocol,
                                    TransportLayer transportLayer,
                                    DelegationTokenCache tokenCache) throws IOException {
-        if (subject == null)
-            throw new IllegalArgumentException("subject cannot be null");
         this.connectionId = connectionId;
-        this.jaasContext = jaasContext;
-        this.subject = subject;
+        this.jaasContexts = jaasContexts;
+        this.subjects = subjects;
         this.credentialCache = credentialCache;
         this.listenerName = listenerName;
         this.securityProtocol = securityProtocol;
@@ -154,6 +152,12 @@ public class SaslServerAuthenticator implements Authenticator {
         if (enabledMechanisms == null || enabledMechanisms.isEmpty())
             throw new IllegalArgumentException("No SASL mechanisms are enabled");
         this.enabledMechanisms = new HashSet<>(enabledMechanisms);
+        for (String mechanism : enabledMechanisms) {
+            if (!jaasContexts.containsKey(mechanism))
+                throw new IllegalArgumentException("Jaas context not specified for SASL mechanism " + mechanism);
+            if (!subjects.containsKey(mechanism))
+                throw new IllegalArgumentException("Subject cannot be null for SASL mechanism " + mechanism);
+        }
 
         // Note that the old principal builder does not support SASL, so we do not need to pass the
         // authenticator or the transport layer
@@ -162,8 +166,9 @@ public class SaslServerAuthenticator implements Authenticator {
 
     private void createSaslServer(String mechanism) throws IOException {
         this.saslMechanism = mechanism;
+        Subject subject = subjects.get(mechanism);
         if (!ScramMechanism.isScram(mechanism))
-            callbackHandler = new SaslServerCallbackHandler(jaasContext);
+            callbackHandler = new SaslServerCallbackHandler(jaasContexts.get(mechanism));
         else
             callbackHandler = new ScramServerCallbackHandler(credentialCache.cache(mechanism, ScramCredential.class), tokenCache);
         callbackHandler.configure(configs, Mode.SERVER, subject, saslMechanism);
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java
index 8d80542..b4875d6 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java
@@ -76,9 +76,9 @@ public final class ScramCredentialUtils {
         return props;
     }
 
-    public static void createCache(CredentialCache cache, Collection<String> enabledMechanisms) {
+    public static void createCache(CredentialCache cache, Collection<String> mechanisms) {
         for (String mechanism : ScramMechanism.mechanismNames()) {
-            if (enabledMechanisms.contains(mechanism))
+            if (mechanisms.contains(mechanism))
                 cache.createCache(mechanism, ScramCredential.class);
         }
     }
diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
index 0d1fbf9..c54260d 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
@@ -144,14 +144,13 @@ public class SslFactory implements Reconfigurable {
     }
 
     @Override
-    public boolean validateReconfiguration(Map<String, ?> configs) {
+    public void validateReconfiguration(Map<String, ?> configs) {
         try {
             SecurityStore newKeystore = maybeCreateNewKeystore(configs);
             if (newKeystore != null)
                 createSSLContext(newKeystore);
-            return true;
         } catch (Exception e) {
-            throw new KafkaException("Validation of dynamic config update failed", e);
+            throw new ConfigException("Validation of dynamic config update failed", e);
         }
     }
 
@@ -163,7 +162,7 @@ public class SslFactory implements Reconfigurable {
                 this.sslContext = createSSLContext(newKeystore);
                 this.keystore = newKeystore;
             } catch (Exception e) {
-                throw new KafkaException("Reconfiguration of SSL keystore failed", e);
+                throw new ConfigException("Reconfiguration of SSL keystore failed", e);
             }
         }
     }
diff --git a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
index 074df10..071deed 100644
--- a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.common.config;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.common.metrics.FakeMetricsReporter;
 import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.MetricsReporter;
@@ -140,6 +141,54 @@ public class AbstractConfigTest {
     }
 
     @Test
+    public void testValuesWithSecondaryPrefix() {
+        String prefix = "listener.name.listener1.";
+        Password saslJaasConfig1 =  new Password("test.myLoginModule1 required;");
+        Password saslJaasConfig2 =  new Password("test.myLoginModule2 required;");
+        Password saslJaasConfig3 =  new Password("test.myLoginModule3 required;");
+        Properties props = new Properties();
+        props.put("listener.name.listener1.test-mechanism.sasl.jaas.config", saslJaasConfig1.value());
+        props.put("test-mechanism.sasl.jaas.config", saslJaasConfig2.value());
+        props.put("sasl.jaas.config", saslJaasConfig3.value());
+        props.put("listener.name.listener1.gssapi.sasl.kerberos.kinit.cmd", "/usr/bin/kinit2");
+        props.put("listener.name.listener1.gssapi.sasl.kerberos.service.name", "testkafka");
+        props.put("listener.name.listener1.gssapi.sasl.kerberos.min.time.before.relogin", "60000");
+        props.put("ssl.provider", "TEST");
+        TestSecurityConfig config = new TestSecurityConfig(props);
+        Map<String, Object> valuesWithPrefixOverride = config.valuesWithPrefixOverride(prefix);
+
+        // prefix with mechanism overrides global
+        assertTrue(config.unused().contains("listener.name.listener1.test-mechanism.sasl.jaas.config"));
+        assertTrue(config.unused().contains("test-mechanism.sasl.jaas.config"));
+        assertEquals(saslJaasConfig1, valuesWithPrefixOverride.get("test-mechanism.sasl.jaas.config"));
+        assertEquals(saslJaasConfig3, valuesWithPrefixOverride.get("sasl.jaas.config"));
+        assertFalse(config.unused().contains("listener.name.listener1.test-mechanism.sasl.jaas.config"));
+        assertFalse(config.unused().contains("test-mechanism.sasl.jaas.config"));
+        assertFalse(config.unused().contains("sasl.jaas.config"));
+
+        // prefix with mechanism overrides default
+        assertFalse(config.unused().contains("sasl.kerberos.kinit.cmd"));
+        assertTrue(config.unused().contains("listener.name.listener1.gssapi.sasl.kerberos.kinit.cmd"));
+        assertFalse(config.unused().contains("gssapi.sasl.kerberos.kinit.cmd"));
+        assertFalse(config.unused().contains("sasl.kerberos.kinit.cmd"));
+        assertEquals("/usr/bin/kinit2", valuesWithPrefixOverride.get("gssapi.sasl.kerberos.kinit.cmd"));
+        assertFalse(config.unused().contains("listener.name.listener1.sasl.kerberos.kinit.cmd"));
+
+        // prefix override for mechanism with no default
+        assertFalse(config.unused().contains("sasl.kerberos.service.name"));
+        assertTrue(config.unused().contains("listener.name.listener1.gssapi.sasl.kerberos.service.name"));
+        assertFalse(config.unused().contains("gssapi.sasl.kerberos.service.name"));
+        assertFalse(config.unused().contains("sasl.kerberos.service.name"));
+        assertEquals("testkafka", valuesWithPrefixOverride.get("gssapi.sasl.kerberos.service.name"));
+        assertFalse(config.unused().contains("listener.name.listener1.gssapi.sasl.kerberos.service.name"));
+
+        // unset with no default
+        assertTrue(config.unused().contains("ssl.provider"));
+        assertNull(valuesWithPrefixOverride.get("gssapi.ssl.provider"));
+        assertTrue(config.unused().contains("ssl.provider"));
+    }
+
+    @Test
     public void testValuesWithPrefixAllOrNothing() {
         String prefix1 = "prefix1.";
         String prefix2 = "prefix2.";
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java b/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
index 86d26d4..6072bf5 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
@@ -26,9 +26,10 @@ import org.junit.Test;
 
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Map;
 
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 public class SaslChannelBuilderTest {
@@ -37,20 +38,20 @@ public class SaslChannelBuilderTest {
     public void testCloseBeforeConfigureIsIdempotent() {
         SaslChannelBuilder builder = createChannelBuilder(SecurityProtocol.SASL_PLAINTEXT);
         builder.close();
-        assertNull(builder.loginManager());
+        assertTrue(builder.loginManagers().isEmpty());
         builder.close();
-        assertNull(builder.loginManager());
+        assertTrue(builder.loginManagers().isEmpty());
     }
 
     @Test
     public void testCloseAfterConfigIsIdempotent() {
         SaslChannelBuilder builder = createChannelBuilder(SecurityProtocol.SASL_PLAINTEXT);
         builder.configure(new HashMap<String, Object>());
-        assertNotNull(builder.loginManager());
+        assertNotNull(builder.loginManagers().get("PLAIN"));
         builder.close();
-        assertNull(builder.loginManager());
+        assertTrue(builder.loginManagers().isEmpty());
         builder.close();
-        assertNull(builder.loginManager());
+        assertTrue(builder.loginManagers().isEmpty());
     }
 
     @Test
@@ -61,17 +62,18 @@ public class SaslChannelBuilderTest {
             builder.configure(Collections.singletonMap(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, "1"));
             fail("Exception should have been thrown");
         } catch (KafkaException e) {
-            assertNull(builder.loginManager());
+            assertTrue(builder.loginManagers().isEmpty());
         }
         builder.close();
-        assertNull(builder.loginManager());
+        assertTrue(builder.loginManagers().isEmpty());
     }
 
     private SaslChannelBuilder createChannelBuilder(SecurityProtocol securityProtocol) {
         TestJaasConfig jaasConfig = new TestJaasConfig();
         jaasConfig.addEntry("jaasContext", PlainLoginModule.class.getName(), new HashMap<String, Object>());
         JaasContext jaasContext = new JaasContext("jaasContext", JaasContext.Type.SERVER, jaasConfig);
-        return new SaslChannelBuilder(Mode.CLIENT, jaasContext, securityProtocol, new ListenerName("PLAIN"),
+        Map<String, JaasContext> jaasContexts = Collections.singletonMap("PLAIN", jaasContext);
+        return new SaslChannelBuilder(Mode.CLIENT, jaasContexts, securityProtocol, new ListenerName("PLAIN"),
                 false, "PLAIN", true, null, null);
     }
 
diff --git a/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java b/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java
index 49d5a86..e8535d2 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java
@@ -188,8 +188,8 @@ public class JaasContextTest {
                 "KafkaServer { test.LoginModuleDefault required; };",
                 "plaintext.KafkaServer { test.LoginModuleOverride requisite; };"
         ));
-        JaasContext context = JaasContext.load(JaasContext.Type.SERVER, new ListenerName("plaintext"),
-                Collections.<String, Object>emptyMap());
+        JaasContext context = JaasContext.loadServerContext(new ListenerName("plaintext"),
+                "SOME-MECHANISM", Collections.<String, Object>emptyMap());
         assertEquals("plaintext.KafkaServer", context.name());
         assertEquals(JaasContext.Type.SERVER, context.type());
         assertEquals(1, context.configurationEntries().size());
@@ -203,8 +203,8 @@ public class JaasContextTest {
                 "KafkaServer { test.LoginModule required; };",
                 "other.KafkaServer { test.LoginModuleOther requisite; };"
         ));
-        JaasContext context = JaasContext.load(JaasContext.Type.SERVER, new ListenerName("plaintext"),
-                Collections.<String, Object>emptyMap());
+        JaasContext context = JaasContext.loadServerContext(new ListenerName("plaintext"),
+                "SOME-MECHANISM", Collections.<String, Object>emptyMap());
         assertEquals("KafkaServer", context.name());
         assertEquals(JaasContext.Type.SERVER, context.type());
         assertEquals(1, context.configurationEntries().size());
@@ -215,24 +215,13 @@ public class JaasContextTest {
     @Test(expected = IllegalArgumentException.class)
     public void testLoadForServerWithWrongListenerName() throws IOException {
         writeConfiguration("Server", "test.LoginModule required;");
-        JaasContext.load(JaasContext.Type.SERVER, new ListenerName("plaintext"),
-                Collections.<String, Object>emptyMap());
-    }
-
-    /**
-     * ListenerName can only be used with Type.SERVER.
-     */
-    @Test(expected = IllegalArgumentException.class)
-    public void testLoadForClientWithListenerName() {
-        JaasContext.load(JaasContext.Type.CLIENT, new ListenerName("foo"),
+        JaasContext.loadServerContext(new ListenerName("plaintext"), "SOME-MECHANISM",
                 Collections.<String, Object>emptyMap());
     }
 
     private AppConfigurationEntry configurationEntry(JaasContext.Type contextType, String jaasConfigProp) {
-        Map<String, Object> configs = new HashMap<>();
-        if (jaasConfigProp != null)
-            configs.put(SaslConfigs.SASL_JAAS_CONFIG, new Password(jaasConfigProp));
-        JaasContext context = JaasContext.load(contextType, null, contextType.name(), configs);
+        Password saslJaasConfig = jaasConfigProp == null ? null : new Password(jaasConfigProp);
+        JaasContext context = JaasContext.load(contextType, null, contextType.name(), saslJaasConfig);
         List<AppConfigurationEntry> entries = context.configurationEntries();
         assertEquals(1, entries.size());
         return entries.get(0);
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
index e3d6b7a..ef2a075 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
@@ -715,7 +715,7 @@ public class SaslAuthenticatorTest {
      * property override is used during authentication.
      */
     @Test
-    public void testDynamicJaasConfiguration() throws Exception {
+    public void testClientDynamicJaasConfiguration() throws Exception {
         SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
         saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         saslServerConfigs.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, Arrays.asList("PLAIN"));
@@ -756,6 +756,37 @@ public class SaslAuthenticatorTest {
         }
     }
 
+    /**
+     * Tests dynamic JAAS configuration property for SASL server. Invalid server credentials
+     * are set in the static JVM-wide configuration instance to ensure that the dynamic
+     * property override is used during authentication.
+     */
+    @Test
+    public void testServerDynamicJaasConfiguration() throws Exception {
+        SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
+        saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+        saslServerConfigs.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, Arrays.asList("PLAIN"));
+        Map<String, Object> serverOptions = new HashMap<>();
+        serverOptions.put("user_user1", "user1-secret");
+        serverOptions.put("user_user2", "user2-secret");
+        saslServerConfigs.put("listener.name.sasl_ssl.plain." + SaslConfigs.SASL_JAAS_CONFIG,
+                TestJaasConfig.jaasConfigProperty("PLAIN", serverOptions));
+        TestJaasConfig staticJaasConfig = new TestJaasConfig();
+        staticJaasConfig.createOrUpdateEntry(TestJaasConfig.LOGIN_CONTEXT_SERVER, PlainLoginModule.class.getName(),
+                Collections.<String, Object>emptyMap());
+        staticJaasConfig.setClientOptions("PLAIN", "user1", "user1-secret");
+        Configuration.setConfiguration(staticJaasConfig);
+        server = createEchoServer(securityProtocol);
+
+        // Check that 'user1' can connect with static Jaas config
+        createAndCheckClientConnection(securityProtocol, "1");
+
+        // Check that user 'user2' can also connect with a Jaas config override
+        saslClientConfigs.put(SaslConfigs.SASL_JAAS_CONFIG,
+                TestJaasConfig.jaasConfigProperty("PLAIN", "user2", "user2-secret"));
+        createAndCheckClientConnection(securityProtocol, "2");
+    }
+
     @Test
     public void testJaasConfigurationForListener() throws Exception {
         SecurityProtocol securityProtocol = SecurityProtocol.SASL_PLAINTEXT;
@@ -986,18 +1017,19 @@ public class SaslAuthenticatorTest {
             throws Exception {
         final ListenerName listenerName = ListenerName.forSecurityProtocol(securityProtocol);
         final Map<String, ?> configs = Collections.emptyMap();
-        final JaasContext jaasContext = JaasContext.load(JaasContext.Type.SERVER, listenerName, configs);
+        final JaasContext jaasContext = JaasContext.loadServerContext(listenerName, saslMechanism, configs);
+        final Map<String, JaasContext> jaasContexts = Collections.singletonMap(saslMechanism, jaasContext);
 
         boolean isScram = ScramMechanism.isScram(saslMechanism);
         if (isScram)
             ScramCredentialUtils.createCache(credentialCache, Arrays.asList(saslMechanism));
-        SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(Mode.SERVER, jaasContext,
+        SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(Mode.SERVER, jaasContexts,
                 securityProtocol, listenerName, false, saslMechanism, true, credentialCache, null) {
 
             @Override
             protected SaslServerAuthenticator buildServerAuthenticator(Map<String, ?> configs, String id,
-                            TransportLayer transportLayer, Subject subject) throws IOException {
-                return new SaslServerAuthenticator(configs, id, jaasContext, subject, null,
+                            TransportLayer transportLayer, Map<String, Subject> subjects) throws IOException {
+                return new SaslServerAuthenticator(configs, id, jaasContexts, subjects, null,
                                 credentialCache, listenerName, securityProtocol, transportLayer, null) {
 
                     @Override
@@ -1032,8 +1064,10 @@ public class SaslAuthenticatorTest {
 
         final ListenerName listenerName = ListenerName.forSecurityProtocol(securityProtocol);
         final Map<String, ?> configs = Collections.emptyMap();
-        final JaasContext jaasContext = JaasContext.load(JaasContext.Type.CLIENT, null, configs);
-        SaslChannelBuilder clientChannelBuilder = new SaslChannelBuilder(Mode.CLIENT, jaasContext,
+        final JaasContext jaasContext = JaasContext.loadClientContext(configs);
+        final Map<String, JaasContext> jaasContexts = Collections.singletonMap(saslMechanism, jaasContext);
+
+        SaslChannelBuilder clientChannelBuilder = new SaslChannelBuilder(Mode.CLIENT, jaasContexts,
                 securityProtocol, listenerName, false, saslMechanism, true, null, null) {
 
             @Override
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java
index 51ea58e..72c2969 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java
@@ -51,7 +51,7 @@ public class SaslServerAuthenticatorTest {
         TransportLayer transportLayer = EasyMock.mock(TransportLayer.class);
         Map<String, ?> configs = Collections.singletonMap(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG,
                 Collections.singletonList(SCRAM_SHA_256.mechanismName()));
-        SaslServerAuthenticator authenticator = setupAuthenticator(configs, transportLayer);
+        SaslServerAuthenticator authenticator = setupAuthenticator(configs, transportLayer, SCRAM_SHA_256.mechanismName());
 
         final Capture<ByteBuffer> size = EasyMock.newCapture();
         EasyMock.expect(transportLayer.read(EasyMock.capture(size))).andAnswer(new IAnswer<Integer>() {
@@ -72,7 +72,7 @@ public class SaslServerAuthenticatorTest {
         TransportLayer transportLayer = EasyMock.mock(TransportLayer.class);
         Map<String, ?> configs = Collections.singletonMap(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG,
                 Collections.singletonList(SCRAM_SHA_256.mechanismName()));
-        SaslServerAuthenticator authenticator = setupAuthenticator(configs, transportLayer);
+        SaslServerAuthenticator authenticator = setupAuthenticator(configs, transportLayer, SCRAM_SHA_256.mechanismName());
 
         final RequestHeader header = new RequestHeader(ApiKeys.METADATA, (short) 0, "clientId", 13243);
         final Struct headerStruct = header.toStruct();
@@ -106,12 +106,13 @@ public class SaslServerAuthenticatorTest {
         }
     }
 
-    private SaslServerAuthenticator setupAuthenticator(Map<String, ?> configs, TransportLayer transportLayer) throws IOException {
+    private SaslServerAuthenticator setupAuthenticator(Map<String, ?> configs, TransportLayer transportLayer, String mechanism) throws IOException {
         TestJaasConfig jaasConfig = new TestJaasConfig();
         jaasConfig.addEntry("jaasContext", PlainLoginModule.class.getName(), new HashMap<String, Object>());
-        JaasContext jaasContext = new JaasContext("jaasContext", JaasContext.Type.SERVER, jaasConfig);
-        Subject subject = new Subject();
-        return new SaslServerAuthenticator(configs, "node", jaasContext, subject, null, new CredentialCache(),
+        Map<String, JaasContext> jaasContexts = Collections.singletonMap(mechanism,
+                new JaasContext("jaasContext", JaasContext.Type.SERVER, jaasConfig));
+        Map<String, Subject> subjects = Collections.singletonMap(mechanism, new Subject());
+        return new SaslServerAuthenticator(configs, "node", jaasContexts, subjects, null, new CredentialCache(),
                 new ListenerName("ssl"), SecurityProtocol.SASL_SSL, transportLayer, new DelegationTokenCache(ScramMechanism.mechanismNames()));
     }
 
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java
index 5336fd7..dafa79d 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java
@@ -54,6 +54,20 @@ public class TestJaasConfig extends Configuration {
         return new Password(loginModule(mechanism) + " required username=" + username + " password=" + password + ";");
     }
 
+    public static Password jaasConfigProperty(String mechanism, Map<String, Object> options) {
+        StringBuilder builder = new StringBuilder();
+        builder.append(loginModule(mechanism));
+        builder.append(" required");
+        for (Map.Entry<String, Object> option : options.entrySet()) {
+            builder.append(' ');
+            builder.append(option.getKey());
+            builder.append('=');
+            builder.append(option.getValue());
+        }
+        builder.append(';');
+        return new Password(builder.toString());
+    }
+
     public void setClientOptions(String saslMechanism, String clientUsername, String clientPassword) {
         Map<String, Object> options = new HashMap<>();
         if (clientUsername != null)
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index a8e1249..f36fc79 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -54,11 +54,13 @@ object KafkaController extends Logging {
 
 }
 
-class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Time, metrics: Metrics, brokerInfo: BrokerInfo,
+class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Time, metrics: Metrics, initialBrokerInfo: BrokerInfo,
                       tokenManager: DelegationTokenManager, threadNamePrefix: Option[String] = None) extends Logging with KafkaMetricsGroup {
 
   this.logIdent = s"[Controller id=${config.brokerId}] "
 
+  @volatile private var brokerInfo = initialBrokerInfo
+
   private val stateChangeLogger = new StateChangeLogger(config.brokerId, inControllerContext = true, None)
   val controllerContext = new ControllerContext
 
@@ -77,6 +79,7 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Ti
 
   private val controllerChangeHandler = new ControllerChangeHandler(this, eventManager)
   private val brokerChangeHandler = new BrokerChangeHandler(this, eventManager)
+  private val brokerModificationsHandlers: mutable.Map[Int, BrokerModificationsHandler] = mutable.Map.empty
   private val topicChangeHandler = new TopicChangeHandler(this, eventManager)
   private val topicDeletionHandler = new TopicDeletionHandler(this, eventManager)
   private val partitionModificationsHandlers: mutable.Map[String, PartitionModificationsHandler] = mutable.Map.empty
@@ -274,6 +277,7 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Ti
     zkClient.unregisterZNodeChangeHandler(partitionReassignmentHandler.path)
     zkClient.unregisterZNodeChangeHandler(preferredReplicaElectionHandler.path)
     zkClient.unregisterZNodeChildChangeHandler(logDirEventNotificationHandler.path)
+    unregisterBrokerModificationsHandler(brokerModificationsHandlers.keySet)
 
     // reset topic deletion manager
     topicDeletionManager.reset()
@@ -360,6 +364,23 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Ti
         s"${newBrokers.mkString(",")}. Signaling restart of topic deletion for these topics")
       topicDeletionManager.resumeDeletionForTopics(replicasForTopicsToBeDeleted.map(_.topic))
     }
+    registerBrokerModificationsHandler(newBrokers)
+  }
+
+  private def registerBrokerModificationsHandler(brokerIds: Iterable[Int]): Unit = {
+    debug(s"Register BrokerModifications handler for $brokerIds")
+    brokerIds.foreach { brokerId =>
+      val brokerModificationsHandler = new BrokerModificationsHandler(this, eventManager, brokerId)
+      zkClient.registerZNodeChangeHandler(brokerModificationsHandler)
+      brokerModificationsHandlers.put(brokerId, brokerModificationsHandler)
+    }
+  }
+
+  private def unregisterBrokerModificationsHandler(brokerIds: Iterable[Int]): Unit = {
+    debug(s"Unregister BrokerModifications handler for $brokerIds")
+    brokerIds.foreach { brokerId =>
+      brokerModificationsHandlers.remove(brokerId).foreach(handler => zkClient.unregisterZNodeChangeHandler(handler.path))
+    }
   }
 
   /*
@@ -374,6 +395,13 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Ti
     info(s"Removed $deadBrokersThatWereShuttingDown from list of shutting down brokers.")
     val allReplicasOnDeadBrokers = controllerContext.replicasOnBrokers(deadBrokers.toSet)
     onReplicasBecomeOffline(allReplicasOnDeadBrokers)
+
+    unregisterBrokerModificationsHandler(deadBrokers)
+  }
+
+  private def onBrokerUpdate(updatedBrokers: Seq[Int]) {
+    info(s"Broker info update callback for ${updatedBrokers.mkString(",")}")
+    sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
   }
 
   /**
@@ -613,6 +641,8 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Ti
     controllerContext.partitionReplicaAssignment = mutable.Map.empty ++ zkClient.getReplicaAssignmentForTopics(controllerContext.allTopics.toSet)
     controllerContext.partitionLeadershipInfo = new mutable.HashMap[TopicPartition, LeaderIsrAndControllerEpoch]
     controllerContext.shuttingDownBrokerIds = mutable.Set.empty[Int]
+    // register broker modifications handlers
+    registerBrokerModificationsHandler(controllerContext.liveBrokers.map(_.id))
     // update the leader and isr cache for all existing partitions from Zookeeper
     updateLeaderAndIsrCache()
     // start the channel manager
@@ -1209,6 +1239,29 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Ti
     }
   }
 
+  case object BrokerModifications extends ControllerEvent {
+    override def state: ControllerState = ControllerState.BrokerChange
+
+    override def process(): Unit = {
+      if (!isActive) return
+      val curBrokers = zkClient.getAllBrokersInCluster.toSet
+      val updatedBrokers = controllerContext.liveBrokers.filter { broker =>
+        val existingBroker = curBrokers.find(_.id == broker.id)
+        existingBroker match {
+          case Some(b) => broker.endPoints != b.endPoints
+          case None => false
+        }
+      }
+      if (updatedBrokers.nonEmpty) {
+        val updatedBrokerIdsSorted = updatedBrokers.map(_.id).toSeq.sorted
+        info(s"Updated brokers: $updatedBrokers")
+
+        controllerContext.liveBrokers = curBrokers // Update broker metadata
+        onBrokerUpdate(updatedBrokerIdsSorted)
+      }
+    }
+  }
+
   case object TopicChange extends ControllerEvent {
     override def state: ControllerState = ControllerState.TopicChange
 
@@ -1458,7 +1511,17 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient, time: Ti
 class BrokerChangeHandler(controller: KafkaController, eventManager: ControllerEventManager) extends ZNodeChildChangeHandler {
   override val path: String = BrokerIdsZNode.path
 
-  override def handleChildChange(): Unit = eventManager.put(controller.BrokerChange)
+  override def handleChildChange(): Unit = {
+    eventManager.put(controller.BrokerChange)
+  }
+}
+
+class BrokerModificationsHandler(controller: KafkaController, eventManager: ControllerEventManager, brokerId: Int) extends ZNodeChangeHandler {
+  override val path: String = BrokerIdZNode.path(brokerId)
+
+  override def handleDataChange(): Unit = {
+    eventManager.put(controller.BrokerModifications)
+  }
 }
 
 class TopicChangeHandler(controller: KafkaController, eventManager: ControllerEventManager) extends ZNodeChildChangeHandler {
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index f6ec974..c11ebcb 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -54,7 +54,6 @@ import scala.util.control.ControlThrowable
  */
 class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time, val credentialProvider: CredentialProvider) extends Logging with KafkaMetricsGroup {
 
-  private val endpoints = config.listeners.map(l => l.listenerName -> l).toMap
   private val maxQueuedRequests = config.queuedMaxRequests
 
   private val maxConnectionsPerIp = config.maxConnectionsPerIp
@@ -72,7 +71,7 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
   private val processors = new ConcurrentHashMap[Int, Processor]()
   private var nextProcessorId = 0
 
-  private[network] val acceptors = mutable.Map[EndPoint, Acceptor]()
+  private[network] val acceptors = new ConcurrentHashMap[EndPoint, Acceptor]()
   private var connectionQuotas: ConnectionQuotas = _
   private var stoppedProcessingRequests = false
 
@@ -82,7 +81,7 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
   def startup() {
     this.synchronized {
       connectionQuotas = new ConnectionQuotas(maxConnectionsPerIp, maxConnectionsPerIpOverrides)
-      createProcessors(config.numNetworkThreads)
+      createProcessors(config.numNetworkThreads, config.listeners)
     }
 
     newGauge("NetworkProcessorAvgIdlePercent",
@@ -111,14 +110,17 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
     info("Started " + acceptors.size + " acceptor threads")
   }
 
-  private def createProcessors(newProcessorsPerListener: Int): Unit = synchronized {
+  private def endpoints = config.listeners.map(l => l.listenerName -> l).toMap
+
+  private def createProcessors(newProcessorsPerListener: Int,
+                               endpoints: Seq[EndPoint]): Unit = synchronized {
 
     val sendBufferSize = config.socketSendBufferBytes
     val recvBufferSize = config.socketReceiveBufferBytes
     val brokerId = config.brokerId
 
     val numProcessorThreads = config.numNetworkThreads
-    config.listeners.foreach { endpoint =>
+    endpoints.foreach { endpoint =>
       val listenerName = endpoint.listenerName
       val securityProtocol = endpoint.securityProtocol
       val listenerProcessors = new ArrayBuffer[Processor]()
@@ -130,12 +132,10 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
       }
       listenerProcessors.foreach(p => processors.put(p.id, p))
 
-      val acceptor = acceptors.getOrElseUpdate(endpoint, {
-        val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, connectionQuotas)
-        KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()
-        acceptor.awaitStartup()
-        acceptor
-      })
+      val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, connectionQuotas)
+      KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()
+      acceptor.awaitStartup()
+      acceptors.put(endpoint, acceptor)
       acceptor.addProcessors(listenerProcessors)
     }
   }
@@ -153,7 +153,7 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
   def stopProcessingRequests() = {
     info("Stopping socket server request processors")
     this.synchronized {
-      acceptors.values.foreach(_.shutdown)
+      acceptors.asScala.values.foreach(_.shutdown)
       processors.asScala.values.foreach(_.shutdown)
       requestChannel.clear()
       stoppedProcessingRequests = true
@@ -161,12 +161,12 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
     info("Stopped socket server request processors")
   }
 
-  def resizeThreadPool(oldNumNetworkThreads: Int, newNumNetworkThreads: Int): Unit = {
+  def resizeThreadPool(oldNumNetworkThreads: Int, newNumNetworkThreads: Int): Unit = synchronized {
     info(s"Resizing network thread pool size for each listener from $oldNumNetworkThreads to $newNumNetworkThreads")
     if (newNumNetworkThreads > oldNumNetworkThreads)
-      createProcessors(newNumNetworkThreads - oldNumNetworkThreads)
+      createProcessors(newNumNetworkThreads - oldNumNetworkThreads, config.listeners)
     else if (newNumNetworkThreads < oldNumNetworkThreads)
-      acceptors.values.foreach(_.removeProcessors(oldNumNetworkThreads - newNumNetworkThreads, requestChannel))
+      acceptors.asScala.values.foreach(_.removeProcessors(oldNumNetworkThreads - newNumNetworkThreads, requestChannel))
   }
 
   /**
@@ -185,9 +185,22 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
 
   def boundPort(listenerName: ListenerName): Int = {
     try {
-      acceptors(endpoints(listenerName)).serverChannel.socket.getLocalPort
+      acceptors.get(endpoints(listenerName)).serverChannel.socket.getLocalPort
     } catch {
-      case e: Exception => throw new KafkaException("Tried to check server's port before server was started or checked for port of non-existing protocol", e)
+      case e: Exception =>
+        throw new KafkaException("Tried to check server's port before server was started or checked for port of non-existing protocol", e)
+    }
+  }
+
+  def addListeners(listenersAdded: Seq[EndPoint]): Unit = synchronized {
+    info(s"Adding listeners for endpoints $listenersAdded")
+    createProcessors(config.numNetworkThreads, listenersAdded)
+  }
+
+  def removeListeners(listenersRemoved: Seq[EndPoint]): Unit = synchronized {
+    info(s"Removing listeners for endpoints $listenersRemoved")
+    listenersRemoved.foreach { endpoint =>
+      acceptors.asScala.remove(endpoint).foreach(_.shutdown())
     }
   }
 
@@ -239,8 +252,8 @@ private[kafka] abstract class AbstractServerThread(connectionQuotas: ConnectionQ
    * Initiates a graceful shutdown by signaling to stop and waiting for the shutdown to complete
    */
   def shutdown(): Unit = {
-    alive.set(false)
-    wakeup()
+    if (alive.getAndSet(false))
+      wakeup()
     shutdownLatch.await()
   }
 
@@ -312,6 +325,11 @@ private[kafka] class Acceptor(val endPoint: EndPoint,
     toRemove.foreach(processor => requestChannel.removeProcessor(processor.id))
   }
 
+  override def shutdown(): Unit = {
+    super.shutdown()
+    processors.foreach(_.shutdown())
+  }
+
   /**
    * Accept loop that checks for new connection attempts
    */
diff --git a/core/src/main/scala/kafka/security/CredentialProvider.scala b/core/src/main/scala/kafka/security/CredentialProvider.scala
index 120e8f9..0e7ebb6 100644
--- a/core/src/main/scala/kafka/security/CredentialProvider.scala
+++ b/core/src/main/scala/kafka/security/CredentialProvider.scala
@@ -17,7 +17,7 @@
 
 package kafka.security
 
-import java.util.{List, Properties}
+import java.util.{Collection, Properties}
 
 import org.apache.kafka.common.security.authenticator.CredentialCache
 import org.apache.kafka.common.security.scram.{ScramCredential, ScramCredentialUtils, ScramMechanism}
@@ -25,10 +25,10 @@ import org.apache.kafka.common.config.ConfigDef
 import org.apache.kafka.common.config.ConfigDef._
 import org.apache.kafka.common.security.token.delegation.DelegationTokenCache
 
-class CredentialProvider(saslEnabledMechanisms: List[String], val tokenCache: DelegationTokenCache) {
+class CredentialProvider(scramMechanisms: Collection[String], val tokenCache: DelegationTokenCache) {
 
   val credentialCache = new CredentialCache
-  ScramCredentialUtils.createCache(credentialCache, saslEnabledMechanisms)
+  ScramCredentialUtils.createCache(credentialCache, scramMechanisms)
 
   def updateCredentials(username: String, config: Properties) {
     for (mechanism <- ScramMechanism.values()) {
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 9c85f9b..a95de0a 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -17,20 +17,22 @@
 
 package kafka.server
 
-import java.nio.charset.StandardCharsets
 import java.util
 import java.util.{Collections, Properties}
 import java.util.concurrent.locks.ReentrantReadWriteLock
 
+import kafka.cluster.EndPoint
 import kafka.log.{LogCleaner, LogConfig, LogManager}
 import kafka.server.DynamicBrokerConfig._
-import kafka.utils.{CoreUtils, Logging}
+import kafka.utils.{CoreUtils, Logging, PasswordEncoder}
 import kafka.zk.{AdminZkClient, KafkaZkClient}
 import org.apache.kafka.common.Reconfigurable
 import org.apache.kafka.common.config.{ConfigDef, ConfigException, SslConfigs}
-import org.apache.kafka.common.network.ListenerReconfigurable
 import org.apache.kafka.common.metrics.MetricsReporter
-import org.apache.kafka.common.utils.{Base64, Utils}
+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 scala.collection._
 import scala.collection.JavaConverters._
@@ -71,11 +73,7 @@ import scala.collection.JavaConverters._
   */
 object DynamicBrokerConfig {
 
-  private val DynamicPasswordConfigs = Set(
-    SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,
-    SslConfigs.SSL_KEY_PASSWORD_CONFIG
-  )
-  private val DynamicSecurityConfigs = SslConfigs.RECONFIGURABLE_CONFIGS.asScala
+  private[server] val DynamicSecurityConfigs = SslConfigs.RECONFIGURABLE_CONFIGS.asScala
 
   val AllDynamicConfigs = mutable.Set[String]()
   AllDynamicConfigs ++= DynamicSecurityConfigs
@@ -83,11 +81,17 @@ object DynamicBrokerConfig {
   AllDynamicConfigs ++= DynamicLogConfig.ReconfigurableConfigs
   AllDynamicConfigs ++= DynamicThreadPool.ReconfigurableConfigs
   AllDynamicConfigs ++= Set(KafkaConfig.MetricReporterClassesProp)
+  AllDynamicConfigs ++= DynamicListenerConfig.ReconfigurableConfigs
 
-  private val PerBrokerConfigs = DynamicSecurityConfigs
+  private val PerBrokerConfigs = DynamicSecurityConfigs  ++
+    DynamicListenerConfig.ReconfigurableConfigs
 
   val ListenerConfigRegex = """listener\.name\.[^.]*\.(.*)""".r
 
+  private[server] val DynamicPasswordConfigs = {
+    val passwordConfigs = KafkaConfig.configKeys.filter(_._2.`type` == ConfigDef.Type.PASSWORD).keySet
+    AllDynamicConfigs.intersect(passwordConfigs)
+  }
 
   def brokerConfigSynonyms(name: String, matchListenerOverride: Boolean): List[String] = {
     name match {
@@ -123,11 +127,14 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
   private val brokerReconfigurables = mutable.Buffer[BrokerReconfigurable]()
   private val lock = new ReentrantReadWriteLock
   private var currentConfig = kafkaConfig
+  private val dynamicConfigPasswordEncoder = maybeCreatePasswordEncoder(kafkaConfig.passwordEncoderSecret)
 
   private[server] def initialize(zkClient: KafkaZkClient): Unit = {
     val adminZkClient = new AdminZkClient(zkClient)
     updateDefaultConfig(adminZkClient.fetchEntityConfig(ConfigType.Broker, ConfigEntityName.Default))
-    updateBrokerConfig(kafkaConfig.brokerId, adminZkClient.fetchEntityConfig(ConfigType.Broker, kafkaConfig.brokerId.toString))
+    val props = adminZkClient.fetchEntityConfig(ConfigType.Broker, kafkaConfig.brokerId.toString)
+    val brokerConfig = maybeReEncodePasswords(props, adminZkClient)
+    updateBrokerConfig(kafkaConfig.brokerId, brokerConfig)
   }
 
   def addReconfigurables(kafkaServer: KafkaServer): Unit = {
@@ -136,6 +143,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
       addBrokerReconfigurable(kafkaServer.logManager.cleaner)
     addReconfigurable(new DynamicLogConfig(kafkaServer.logManager))
     addReconfigurable(new DynamicMetricsReporters(kafkaConfig.brokerId, kafkaServer))
+    addBrokerReconfigurable(new DynamicListenerConfig(kafkaServer))
   }
 
   def addReconfigurable(reconfigurable: Reconfigurable): Unit = CoreUtils.inWriteLock(lock) {
@@ -187,22 +195,37 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
     }
   }
 
+  private def maybeCreatePasswordEncoder(secret: Option[Password]): Option[PasswordEncoder] = {
+   secret.map { secret =>
+      new PasswordEncoder(secret,
+        kafkaConfig.passwordEncoderKeyFactoryAlgorithm,
+        kafkaConfig.passwordEncoderCipherAlgorithm,
+        kafkaConfig.passwordEncoderKeyLength,
+        kafkaConfig.passwordEncoderIterations)
+    }
+  }
+
+  private def passwordEncoder: PasswordEncoder = {
+    dynamicConfigPasswordEncoder.getOrElse(throw new ConfigException("Password encoder secret not configured"))
+  }
+
   private[server] def toPersistentProps(configProps: Properties, perBrokerConfig: Boolean): Properties = {
     val props = configProps.clone().asInstanceOf[Properties]
-    // TODO (KAFKA-6246): encrypt passwords
+
     def encodePassword(configName: String): Unit = {
       val value = props.getProperty(configName)
       if (value != null) {
         if (!perBrokerConfig)
           throw new ConfigException("Password config can be defined only at broker level")
-        props.setProperty(configName, Base64.encoder.encodeToString(value.getBytes(StandardCharsets.UTF_8)))
+        props.setProperty(configName, passwordEncoder.encode(new Password(value)))
       }
     }
     DynamicPasswordConfigs.foreach(encodePassword)
     props
   }
 
-  private[server] def fromPersistentProps(persistentProps: Properties, perBrokerConfig: Boolean): Properties = {
+  private[server] def fromPersistentProps(persistentProps: Properties,
+                                          perBrokerConfig: Boolean): Properties = {
     val props = persistentProps.clone().asInstanceOf[Properties]
 
     // Remove all invalid configs from `props`
@@ -219,17 +242,50 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
     if (!perBrokerConfig)
       removeInvalidProps(perBrokerConfigs(props), "Per-broker configs defined at default cluster level will be ignored")
 
-    // TODO (KAFKA-6246): encrypt passwords
     def decodePassword(configName: String): Unit = {
       val value = props.getProperty(configName)
       if (value != null) {
-        props.setProperty(configName, new String(Base64.decoder.decode(value), StandardCharsets.UTF_8))
+        try {
+          props.setProperty(configName, passwordEncoder.decode(value).value)
+        } catch {
+          case e: Exception =>
+            error(s"Dynamic password config $configName could not be decoded, ignoring.", e)
+            props.remove(configName)
+        }
       }
     }
+
     DynamicPasswordConfigs.foreach(decodePassword)
     props
   }
 
+  // If the secret has changed, password.encoder.old.secret contains the old secret that was used
+  // to encode the configs in ZK. Decode passwords using the old secret and update ZK with values
+  // encoded using the current secret. Ignore any errors during decoding since old secret may not
+  // 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)) {
+      maybeCreatePasswordEncoder(kafkaConfig.passwordEncoderOldSecret).foreach { passwordDecoder =>
+        DynamicPasswordConfigs.foreach { configName =>
+          val value = props.getProperty(configName)
+          if (value != null) {
+            val decoded = try {
+              Some(passwordDecoder.decode(value).value)
+            } catch {
+              case _: Exception =>
+                debug(s"Dynamic password config $configName could not be decoded using old secret, new secret will be used.")
+                None
+            }
+            decoded.foreach { value => props.put(configName, passwordEncoder.encode(new Password(value))) }
+          }
+        }
+        adminZkClient.changeBrokerConfig(Seq(kafkaConfig.brokerId), props)
+      }
+    }
+    props
+  }
+
   private[server] def validate(props: Properties, perBrokerConfig: Boolean): Unit = CoreUtils.inReadLock(lock) {
     def checkInvalidProps(invalidPropNames: Set[String], errorMessage: String): Unit = {
       if (invalidPropNames.nonEmpty)
@@ -331,14 +387,18 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
     newProps ++= staticBrokerConfigs
     overrideProps(newProps, dynamicDefaultConfigs)
     overrideProps(newProps, dynamicBrokerConfigs)
-    val newConfig = processReconfiguration(newProps, validateOnly = false)
+    val oldConfig = currentConfig
+    val (newConfig, brokerReconfigurablesToUpdate) = processReconfiguration(newProps, validateOnly = false)
     if (newConfig ne currentConfig) {
       currentConfig = newConfig
-      kafkaConfig.updateCurrentConfig(currentConfig)
+      kafkaConfig.updateCurrentConfig(newConfig)
+
+      // Process BrokerReconfigurable updates after current config is updated
+      brokerReconfigurablesToUpdate.foreach(_.reconfigure(oldConfig, newConfig))
     }
   }
 
-  private def processReconfiguration(newProps: Map[String, String], validateOnly: Boolean): KafkaConfig = {
+  private def processReconfiguration(newProps: Map[String, String], validateOnly: Boolean): (KafkaConfig, List[BrokerReconfigurable]) = {
     val newConfig = new KafkaConfig(newProps.asJava, !validateOnly, None)
     val updatedMap = updatedConfigs(newConfig.originalsFromThisConfig, currentConfig.originals)
     if (updatedMap.nonEmpty) {
@@ -352,16 +412,22 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
             val newValues = newConfig.valuesFromThisConfigWithPrefixOverride(listenerName.configPrefix)
             val updatedKeys = updatedConfigs(newValues, oldValues).keySet
             if (needsReconfiguration(listenerReconfigurable.reconfigurableConfigs, updatedKeys))
-              processReconfigurable(listenerReconfigurable, newValues, customConfigs, validateOnly)
+              processReconfigurable(listenerReconfigurable, updatedKeys, newValues, customConfigs, validateOnly)
           case reconfigurable =>
             if (needsReconfiguration(reconfigurable.reconfigurableConfigs, updatedMap.keySet))
-              processReconfigurable(reconfigurable, newConfig.valuesFromThisConfig, customConfigs, validateOnly)
+              processReconfigurable(reconfigurable, updatedMap.keySet, newConfig.valuesFromThisConfig, customConfigs, validateOnly)
         }
+
+        // BrokerReconfigurable updates are processed after config is updated. Only do the validation here.
+        val brokerReconfigurablesToUpdate = mutable.Buffer[BrokerReconfigurable]()
         brokerReconfigurables.foreach { reconfigurable =>
-          if (needsReconfiguration(reconfigurable.reconfigurableConfigs.asJava, updatedMap.keySet))
-            processBrokerReconfigurable(reconfigurable, currentConfig, newConfig, validateOnly)
+          if (needsReconfiguration(reconfigurable.reconfigurableConfigs.asJava, updatedMap.keySet)) {
+            reconfigurable.validateReconfiguration(newConfig)
+            if (!validateOnly)
+              brokerReconfigurablesToUpdate += reconfigurable
+          }
         }
-        newConfig
+        (newConfig, brokerReconfigurablesToUpdate.toList)
       } catch {
         case e: Exception =>
           if (!validateOnly)
@@ -370,7 +436,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
       }
     }
     else
-      currentConfig
+      (currentConfig, List.empty)
   }
 
   private def needsReconfiguration(reconfigurableConfigs: util.Set[String], updatedKeys: Set[String]): Boolean = {
@@ -378,27 +444,25 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
   }
 
   private def processReconfigurable(reconfigurable: Reconfigurable,
+                                    updatedConfigNames: Set[String],
                                     allNewConfigs: util.Map[String, _],
                                     newCustomConfigs: util.Map[String, Object],
                                     validateOnly: Boolean): Unit = {
     val newConfigs = new util.HashMap[String, Object]
     allNewConfigs.asScala.foreach { case (k, v) => newConfigs.put(k, v.asInstanceOf[AnyRef]) }
     newConfigs.putAll(newCustomConfigs)
-    if (validateOnly) {
-      if (!reconfigurable.validateReconfiguration(newConfigs))
-        throw new ConfigException("Validation of dynamic config update failed")
-    } else
-      reconfigurable.reconfigure(newConfigs)
-  }
+    try {
+      reconfigurable.validateReconfiguration(newConfigs)
+    } catch {
+      case e: ConfigException => throw e
+      case _: Exception =>
+        throw new ConfigException(s"Validation of dynamic config update of $updatedConfigNames failed with class ${reconfigurable.getClass}")
+    }
 
-  private def processBrokerReconfigurable(reconfigurable: BrokerReconfigurable,
-                                          oldConfig: KafkaConfig,
-                                          newConfig: KafkaConfig,
-                                          validateOnly: Boolean): Unit = {
-    if (validateOnly)
-      reconfigurable.validateReconfiguration(newConfig)
-    else
-      reconfigurable.reconfigure(oldConfig, newConfig)
+    if (!validateOnly) {
+      info(s"Reconfiguring $reconfigurable, updated configs: $updatedConfigNames custom configs: $newCustomConfigs")
+      reconfigurable.reconfigure(newConfigs)
+    }
   }
 }
 
@@ -427,11 +491,10 @@ class DynamicLogConfig(logManager: LogManager) extends Reconfigurable with Loggi
     DynamicLogConfig.ReconfigurableConfigs.asJava
   }
 
-  override def validateReconfiguration(configs: util.Map[String, _]): Boolean = {
+  override def validateReconfiguration(configs: util.Map[String, _]): Unit = {
     // For update of topic config overrides, only config names and types are validated
     // Names and types have already been validated. For consistency with topic config
     // validation, no additional validation is performed.
-    true
   }
 
   override def reconfigure(configs: util.Map[String, _]): Unit = {
@@ -538,7 +601,7 @@ class DynamicMetricsReporters(brokerId: Int, server: KafkaServer) extends Reconf
     configs
   }
 
-  override def validateReconfiguration(configs: util.Map[String, _]): Boolean = {
+  override def validateReconfiguration(configs: util.Map[String, _]): Unit = {
     val updatedMetricsReporters = metricsReporterClasses(configs)
 
     // Ensure all the reporter classes can be loaded and have a default constructor
@@ -548,10 +611,11 @@ class DynamicMetricsReporters(brokerId: Int, server: KafkaServer) extends Reconf
     }
 
     // Validate the new configuration using every reconfigurable reporter instance that is not being deleted
-    currentReporters.values.forall {
+    currentReporters.values.foreach {
       case reporter: Reconfigurable =>
-        !updatedMetricsReporters.contains(reporter.getClass.getName) || reporter.validateReconfiguration(configs)
-      case _ => true
+        if (updatedMetricsReporters.contains(reporter.getClass.getName))
+          reporter.validateReconfiguration(configs)
+      case _ =>
     }
   }
 
@@ -588,3 +652,103 @@ class DynamicMetricsReporters(brokerId: Int, server: KafkaServer) extends Reconf
     configs.get(KafkaConfig.MetricReporterClassesProp).asInstanceOf[util.List[String]].asScala
   }
 }
+object DynamicListenerConfig {
+
+  val ReconfigurableConfigs = Set(
+    // Listener configs
+    KafkaConfig.AdvertisedListenersProp,
+    KafkaConfig.ListenersProp,
+    KafkaConfig.ListenerSecurityProtocolMapProp,
+
+    // SSL configs
+    KafkaConfig.PrincipalBuilderClassProp,
+    KafkaConfig.SslProtocolProp,
+    KafkaConfig.SslProviderProp,
+    KafkaConfig.SslCipherSuitesProp,
+    KafkaConfig.SslEnabledProtocolsProp,
+    KafkaConfig.SslKeystoreTypeProp,
+    KafkaConfig.SslKeystoreLocationProp,
+    KafkaConfig.SslKeystorePasswordProp,
+    KafkaConfig.SslKeyPasswordProp,
+    KafkaConfig.SslTruststoreTypeProp,
+    KafkaConfig.SslTruststoreLocationProp,
+    KafkaConfig.SslTruststorePasswordProp,
+    KafkaConfig.SslKeyManagerAlgorithmProp,
+    KafkaConfig.SslTrustManagerAlgorithmProp,
+    KafkaConfig.SslEndpointIdentificationAlgorithmProp,
+    KafkaConfig.SslSecureRandomImplementationProp,
+    KafkaConfig.SslClientAuthProp,
+
+    // SASL configs
+    KafkaConfig.SaslMechanismInterBrokerProtocolProp,
+    KafkaConfig.SaslJaasConfigProp,
+    KafkaConfig.SaslEnabledMechanismsProp,
+    KafkaConfig.SaslKerberosServiceNameProp,
+    KafkaConfig.SaslKerberosKinitCmdProp,
+    KafkaConfig.SaslKerberosTicketRenewWindowFactorProp,
+    KafkaConfig.SaslKerberosTicketRenewJitterProp,
+    KafkaConfig.SaslKerberosMinTimeBeforeReloginProp,
+    KafkaConfig.SaslKerberosPrincipalToLocalRulesProp
+  )
+}
+
+class DynamicListenerConfig(server: KafkaServer) extends BrokerReconfigurable with Logging {
+
+  override def reconfigurableConfigs: Set[String] = {
+    DynamicListenerConfig.ReconfigurableConfigs
+  }
+
+  def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+
+    def immutableListenerConfigs(kafkaConfig: KafkaConfig, prefix: String): Map[String, AnyRef] = {
+      newConfig.originals.asScala
+        .filterKeys(_.startsWith(prefix))
+        .filterKeys(k => !DynamicSecurityConfigs.contains(k))
+    }
+
+    val oldConfig = server.config
+    val newListeners = listenersToMap(newConfig.listeners)
+    val newAdvertisedListeners = listenersToMap(newConfig.advertisedListeners)
+    val oldListeners = listenersToMap(oldConfig.listeners)
+    if (!newAdvertisedListeners.keySet.subsetOf(newListeners.keySet))
+      throw new ConfigException(s"Advertised listeners '$newAdvertisedListeners' must be a subset of listeners '$newListeners'")
+    if (newListeners.keySet != newConfig.listenerSecurityProtocolMap.keySet)
+      throw new ConfigException(s"Listeners '$newListeners' and listener map '${newConfig.listenerSecurityProtocolMap}' don't match")
+    newListeners.keySet.intersect(oldListeners.keySet).foreach { listenerName =>
+      val prefix = listenerName.configPrefix
+      val newListenerProps = immutableListenerConfigs(newConfig, prefix)
+      val oldListenerProps = immutableListenerConfigs(oldConfig, prefix)
+      if (newListenerProps != oldListenerProps)
+        throw new ConfigException(s"Configs cannot be updated dynamically for existing listener $listenerName, " +
+          "restart broker or create a new listener for update")
+      if (oldConfig.listenerSecurityProtocolMap(listenerName) != newConfig.listenerSecurityProtocolMap(listenerName))
+        throw new ConfigException(s"Security protocol cannot be updated for existing listener $listenerName")
+    }
+    if (!newAdvertisedListeners.contains(newConfig.interBrokerListenerName))
+      throw new ConfigException(s"Advertised listener must be specified for inter-broker listener ${newConfig.interBrokerListenerName}")
+  }
+
+  def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+    val newListeners = newConfig.listeners
+    val newListenerMap = listenersToMap(newListeners)
+    val oldListeners = oldConfig.listeners
+    val oldListenerMap = listenersToMap(oldListeners)
+    val listenersRemoved = oldListeners.filterNot(e => newListenerMap.contains(e.listenerName))
+    val listenersAdded = newListeners.filterNot(e => oldListenerMap.contains(e.listenerName))
+
+    // Clear SASL login cache to force re-login
+    if (listenersAdded.nonEmpty || listenersRemoved.nonEmpty)
+      LoginManager.closeAll()
+
+    server.socketServer.removeListeners(listenersRemoved)
+    if (listenersAdded.nonEmpty)
+      server.socketServer.addListeners(listenersAdded)
+
+    server.zkClient.updateBrokerInfoInZk(server.createBrokerInfo)
+  }
+
+  private def listenersToMap(listeners: Seq[EndPoint]): Map[ListenerName, EndPoint] =
+    listeners.map(e => (e.listenerName, e)).toMap
+
+}
+
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 2dd6951..1f448af 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -1291,7 +1291,7 @@ class KafkaApis(val requestChannel: RequestChannel,
   }
 
   def handleSaslHandshakeRequest(request: RequestChannel.Request) {
-    sendResponseMaybeThrottle(request, _ => new SaslHandshakeResponse(Errors.ILLEGAL_SASL_STATE, config.saslEnabledMechanisms))
+    sendResponseMaybeThrottle(request, _ => new SaslHandshakeResponse(Errors.ILLEGAL_SASL_STATE, Collections.emptySet()))
   }
 
   def handleSaslAuthenticateRequest(request: RequestChannel.Request) {
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 144dd65..64698f7 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -223,6 +223,11 @@ object Defaults {
   val DelegationTokenMaxLifeTimeMsDefault = 7 * 24 * 60 * 60 * 1000L
   val DelegationTokenExpiryTimeMsDefault = 24 * 60 * 60 * 1000L
   val DelegationTokenExpiryCheckIntervalMsDefault = 1 * 60 * 60 * 1000L
+
+  /** ********* Password encryption configuration for dynamic configs *********/
+  val PasswordEncoderCipherAlgorithm = "AES/CBC/PKCS5Padding"
+  val PasswordEncoderKeyLength = 128
+  val PasswordEncoderIterations = 4096
 }
 
 object KafkaConfig {
@@ -412,6 +417,7 @@ object KafkaConfig {
 
   /** ********* SASL Configuration ****************/
   val SaslMechanismInterBrokerProtocolProp = "sasl.mechanism.inter.broker.protocol"
+  val SaslJaasConfigProp = SaslConfigs.SASL_JAAS_CONFIG
   val SaslEnabledMechanismsProp = BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG
   val SaslKerberosServiceNameProp = SaslConfigs.SASL_KERBEROS_SERVICE_NAME
   val SaslKerberosKinitCmdProp = SaslConfigs.SASL_KERBEROS_KINIT_CMD
@@ -426,6 +432,14 @@ object KafkaConfig {
   val DelegationTokenExpiryTimeMsProp = "delegation.token.expiry.time.ms"
   val DelegationTokenExpiryCheckIntervalMsProp = "delegation.token.expiry.check.interval.ms"
 
+  /** ********* Password encryption configuration for dynamic configs *********/
+  val PasswordEncoderSecretProp = "password.encoder.secret"
+  val PasswordEncoderOldSecretProp = "password.encoder.old.secret"
+  val PasswordEncoderKeyFactoryAlgorithmProp = "password.encoder.keyfactory.algorithm"
+  val PasswordEncoderCipherAlgorithmProp = "password.encoder.cipher.algorithm"
+  val PasswordEncoderKeyLengthProp =  "password.encoder.key.length"
+  val PasswordEncoderIterationsProp =  "password.encoder.iterations"
+
   /* Documentation */
   /** ********* Zookeeper Configuration ***********/
   val ZkConnectDoc = "Zookeeper host string"
@@ -689,6 +703,7 @@ object KafkaConfig {
 
   /** ********* Sasl Configuration ****************/
   val SaslMechanismInterBrokerProtocolDoc = "SASL mechanism used for inter-broker communication. Default is GSSAPI."
+  val SaslJaasConfigDoc = SaslConfigs.SASL_JAAS_CONFIG_DOC
   val SaslEnabledMechanismsDoc = SaslConfigs.SASL_ENABLED_MECHANISMS_DOC
   val SaslKerberosServiceNameDoc = SaslConfigs.SASL_KERBEROS_SERVICE_NAME_DOC
   val SaslKerberosKinitCmdDoc = SaslConfigs.SASL_KERBEROS_KINIT_CMD_DOC
@@ -704,6 +719,16 @@ object KafkaConfig {
   val DelegationTokenExpiryTimeMsDoc = "The token validity time in seconds before the token needs to be renewed. Default value 1 day."
   val DelegationTokenExpiryCheckIntervalDoc = "Scan interval to remove expired delegation tokens."
 
+  /** ********* Password encryption configuration for dynamic configs *********/
+  val PasswordEncoderSecretDoc = "The secret used for encoding dynamically configured passwords for this broker."
+  val PasswordEncoderOldSecretDoc = "The old secret that was used for encoding dynamically configured passwords. " +
+    "This is required only when the secret is updated. If specified, all dynamically encoded passwords are " +
+    s"decoded using this old secret and re-encoded using $PasswordEncoderSecretProp when broker starts up."
+  val PasswordEncoderKeyFactoryAlgorithmDoc = "The SecretKeyFactory algorithm used for encoding dynamically configured passwords. " +
+    "Default is PBKDF2WithHmacSHA512 if available and PBKDF2WithHmacSHA1 otherwise."
+  val PasswordEncoderCipherAlgorithmDoc = "The Cipher algorithm used for encoding dynamically configured passwords."
+  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 = {
     import ConfigDef.Importance._
@@ -898,6 +923,7 @@ object KafkaConfig {
 
       /** ********* Sasl Configuration ****************/
       .define(SaslMechanismInterBrokerProtocolProp, STRING, Defaults.SaslMechanismInterBrokerProtocol, MEDIUM, SaslMechanismInterBrokerProtocolDoc)
+      .define(SaslJaasConfigProp, PASSWORD, null, MEDIUM, SaslJaasConfigDoc)
       .define(SaslEnabledMechanismsProp, LIST, Defaults.SaslEnabledMechanisms, MEDIUM, SaslEnabledMechanismsDoc)
       .define(SaslKerberosServiceNameProp, STRING, null, MEDIUM, SaslKerberosServiceNameDoc)
       .define(SaslKerberosKinitCmdProp, STRING, Defaults.SaslKerberosKinitCmd, MEDIUM, SaslKerberosKinitCmdDoc)
@@ -911,6 +937,13 @@ object KafkaConfig {
       .define(DelegationTokenExpiryTimeMsProp, LONG, Defaults.DelegationTokenExpiryTimeMsDefault, atLeast(1), MEDIUM, DelegationTokenExpiryTimeMsDoc)
       .define(DelegationTokenExpiryCheckIntervalMsProp, LONG, Defaults.DelegationTokenExpiryCheckIntervalMsDefault, atLeast(1), LOW, DelegationTokenExpiryCheckIntervalDoc)
 
+      /** ********* Password encryption configuration for dynamic configs *********/
+      .define(PasswordEncoderSecretProp, PASSWORD, null, MEDIUM, PasswordEncoderSecretDoc)
+      .define(PasswordEncoderOldSecretProp, PASSWORD, null, MEDIUM, PasswordEncoderOldSecretDoc)
+      .define(PasswordEncoderKeyFactoryAlgorithmProp, STRING, null, LOW, PasswordEncoderKeyFactoryAlgorithmDoc)
+      .define(PasswordEncoderCipherAlgorithmProp, STRING, Defaults.PasswordEncoderCipherAlgorithm, LOW, PasswordEncoderCipherAlgorithmDoc)
+      .define(PasswordEncoderKeyLengthProp, INT, Defaults.PasswordEncoderKeyLength, atLeast(8), LOW, PasswordEncoderKeyLengthDoc)
+      .define(PasswordEncoderIterationsProp, INT, Defaults.PasswordEncoderIterations, atLeast(1024), LOW, PasswordEncoderIterationsDoc)
   }
 
   def configNames() = configDef.names().asScala.toList.sorted
@@ -942,9 +975,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
 
   def this(props: java.util.Map[_, _]) = this(props, true, None)
   def this(props: java.util.Map[_, _], doLog: Boolean) = this(props, doLog, None)
-  private[server] val dynamicConfig = dynamicConfigOverride.getOrElse(new DynamicBrokerConfig(this))
   // Cache the current config to avoid acquiring read lock to access from dynamicConfig
   @volatile private var currentConfig = this
+  private[server] val dynamicConfig = dynamicConfigOverride.getOrElse(new DynamicBrokerConfig(this))
 
   private[server] def updateCurrentConfig(newConfig: KafkaConfig): Unit = {
     this.currentConfig = newConfig
@@ -1076,8 +1109,6 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
   val leaderImbalanceCheckIntervalSeconds = getLong(KafkaConfig.LeaderImbalanceCheckIntervalSecondsProp)
   def uncleanLeaderElectionEnable: java.lang.Boolean = getBoolean(KafkaConfig.UncleanLeaderElectionEnableProp)
 
-  val (interBrokerListenerName, interBrokerSecurityProtocol) = getInterBrokerListenerNameAndSecurityProtocol
-
   // We keep the user-provided String as `ApiVersion.apply` can choose a slightly different version (eg if `0.10.0`
   // is passed, `0.10.0-IV0` may be picked)
   val interBrokerProtocolVersionString = getString(KafkaConfig.InterBrokerProtocolVersionProp)
@@ -1120,32 +1151,21 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
   val metricSampleWindowMs = getLong(KafkaConfig.MetricSampleWindowMsProp)
   val metricRecordingLevel = getString(KafkaConfig.MetricRecordingLevelProp)
 
-  /** ********* SSL Configuration **************/
-  val principalBuilderClass = getClass(KafkaConfig.PrincipalBuilderClassProp)
-  val sslProtocol = getString(KafkaConfig.SslProtocolProp)
-  val sslProvider = getString(KafkaConfig.SslProviderProp)
-  val sslEnabledProtocols = getList(KafkaConfig.SslEnabledProtocolsProp)
-  def sslKeystoreType = getString(KafkaConfig.SslKeystoreTypeProp)
-  def sslKeystoreLocation = getString(KafkaConfig.SslKeystoreLocationProp)
-  def sslKeystorePassword = getPassword(KafkaConfig.SslKeystorePasswordProp)
-  def sslKeyPassword = getPassword(KafkaConfig.SslKeyPasswordProp)
-  val sslTruststoreType = getString(KafkaConfig.SslTruststoreTypeProp)
-  val sslTruststoreLocation = getString(KafkaConfig.SslTruststoreLocationProp)
-  val sslTruststorePassword = getPassword(KafkaConfig.SslTruststorePasswordProp)
-  val sslKeyManagerAlgorithm = getString(KafkaConfig.SslKeyManagerAlgorithmProp)
-  val sslTrustManagerAlgorithm = getString(KafkaConfig.SslTrustManagerAlgorithmProp)
-  val sslClientAuth = getString(KafkaConfig.SslClientAuthProp)
-  val sslCipher = getList(KafkaConfig.SslCipherSuitesProp)
-
-  /** ********* Sasl Configuration **************/
-  val saslMechanismInterBrokerProtocol = getString(KafkaConfig.SaslMechanismInterBrokerProtocolProp)
-  val saslEnabledMechanisms = getList(KafkaConfig.SaslEnabledMechanismsProp)
-  val saslKerberosServiceName = getString(KafkaConfig.SaslKerberosServiceNameProp)
-  val saslKerberosKinitCmd = getString(KafkaConfig.SaslKerberosKinitCmdProp)
-  val saslKerberosTicketRenewWindowFactor = getDouble(KafkaConfig.SaslKerberosTicketRenewWindowFactorProp)
-  val saslKerberosTicketRenewJitter = getDouble(KafkaConfig.SaslKerberosTicketRenewJitterProp)
-  val saslKerberosMinTimeBeforeRelogin = getLong(KafkaConfig.SaslKerberosMinTimeBeforeReloginProp)
-  val saslKerberosPrincipalToLocalRules = getList(KafkaConfig.SaslKerberosPrincipalToLocalRulesProp)
+  /** ********* SSL/SASL Configuration **************/
+  // Security configs may be overridden for listeners, so it is not safe to use the base values
+  // Hence the base SSL/SASL configs are not fields of KafkaConfig, listener configs should be
+  // retrieved using KafkaConfig#valuesWithPrefixOverride
+  private def saslEnabledMechanisms(listenerName: ListenerName): Set[String] = {
+    val value = valuesWithPrefixOverride(listenerName.configPrefix).get(KafkaConfig.SaslEnabledMechanismsProp)
+    if (value != null)
+      value.asInstanceOf[util.List[String]].asScala.toSet
+    else
+      Set.empty[String]
+  }
+
+  def interBrokerListenerName = getInterBrokerListenerNameAndSecurityProtocol._1
+  def interBrokerSecurityProtocol = getInterBrokerListenerNameAndSecurityProtocol._2
+  def saslMechanismInterBrokerProtocol = getString(KafkaConfig.SaslMechanismInterBrokerProtocolProp)
   val saslInterBrokerHandshakeRequestEnable = interBrokerProtocolVersion >= KAFKA_0_10_0_IV1
 
   /** ********* DelegationToken Configuration **************/
@@ -1155,6 +1175,14 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
   val delegationTokenExpiryTimeMs = getLong(KafkaConfig.DelegationTokenExpiryTimeMsProp)
   val delegationTokenExpiryCheckIntervalMs = getLong(KafkaConfig.DelegationTokenExpiryCheckIntervalMsProp)
 
+  /** ********* Password encryption configuration for dynamic configs *********/
+  def passwordEncoderSecret = Option(getPassword(KafkaConfig.PasswordEncoderSecretProp))
+  def passwordEncoderOldSecret = Option(getPassword(KafkaConfig.PasswordEncoderOldSecretProp))
+  def passwordEncoderCipherAlgorithm = getString(KafkaConfig.PasswordEncoderCipherAlgorithmProp)
+  def passwordEncoderKeyFactoryAlgorithm = Option(getString(KafkaConfig.PasswordEncoderKeyFactoryAlgorithmProp))
+  def passwordEncoderKeyLength = getInt(KafkaConfig.PasswordEncoderKeyLengthProp)
+  def passwordEncoderIterations = getInt(KafkaConfig.PasswordEncoderIterationsProp)
+
   /** ********* Quota Configuration **************/
   val producerQuotaBytesPerSecondDefault = getLong(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp)
   val consumerQuotaBytesPerSecondDefault = getLong(KafkaConfig.ConsumerQuotaBytesPerSecondDefaultProp)
@@ -1170,9 +1198,6 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
 
   val deleteTopicEnable = getBoolean(KafkaConfig.DeleteTopicEnableProp)
   def compressionType = getString(KafkaConfig.CompressionTypeProp)
-  val listeners: Seq[EndPoint] = getListeners
-  val advertisedListeners: Seq[EndPoint] = getAdvertisedListeners
-  private[kafka] lazy val listenerSecurityProtocolMap = getListenerSecurityProtocolMap
 
   def addReconfigurable(reconfigurable: Reconfigurable): Unit = {
     dynamicConfig.addReconfigurable(reconfigurable)
@@ -1203,7 +1228,7 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
 
   // If the user did not define listeners but did define host or port, let's use them in backward compatible way
   // If none of those are defined, we default to PLAINTEXT://:9092
-  private def getListeners: Seq[EndPoint] = {
+  def listeners: Seq[EndPoint] = {
     Option(getString(KafkaConfig.ListenersProp)).map { listenerProp =>
       CoreUtils.listenerListToEndPoints(listenerProp, listenerSecurityProtocolMap)
     }.getOrElse(CoreUtils.listenerListToEndPoints("PLAINTEXT://" + hostName + ":" + port, listenerSecurityProtocolMap))
@@ -1212,14 +1237,14 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
   // If the user defined advertised listeners, we use those
   // If he didn't but did define advertised host or port, we'll use those and fill in the missing value from regular host / port or defaults
   // If none of these are defined, we'll use the listeners
-  private def getAdvertisedListeners: Seq[EndPoint] = {
+  def advertisedListeners: Seq[EndPoint] = {
     val advertisedListenersProp = getString(KafkaConfig.AdvertisedListenersProp)
     if (advertisedListenersProp != null)
       CoreUtils.listenerListToEndPoints(advertisedListenersProp, listenerSecurityProtocolMap)
     else if (getString(KafkaConfig.AdvertisedHostNameProp) != null || getInt(KafkaConfig.AdvertisedPortProp) != null)
       CoreUtils.listenerListToEndPoints("PLAINTEXT://" + advertisedHostName + ":" + advertisedPort, listenerSecurityProtocolMap)
     else
-      getListeners
+      listeners
   }
 
   private def getInterBrokerListenerNameAndSecurityProtocol: (ListenerName, SecurityProtocol) = {
@@ -1248,7 +1273,7 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
     }
   }
 
-  private def getListenerSecurityProtocolMap: Map[ListenerName, SecurityProtocol] = {
+  def listenerSecurityProtocolMap: Map[ListenerName, SecurityProtocol] = {
     getMap(KafkaConfig.ListenerSecurityProtocolMapProp, getString(KafkaConfig.ListenerSecurityProtocolMapProp))
       .map { case (listenerName, protocolName) =>
       ListenerName.normalised(listenerName) -> getSecurityProtocol(protocolName, KafkaConfig.ListenerSecurityProtocolMapProp)
@@ -1295,7 +1320,7 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
     val interBrokerUsesSasl = interBrokerSecurityProtocol == SecurityProtocol.SASL_PLAINTEXT || interBrokerSecurityProtocol == SecurityProtocol.SASL_SSL
     require(!interBrokerUsesSasl || saslInterBrokerHandshakeRequestEnable || saslMechanismInterBrokerProtocol == SaslConfigs.GSSAPI_MECHANISM,
       s"Only GSSAPI mechanism is supported for inter-broker communication with SASL when inter.broker.protocol.version is set to $interBrokerProtocolVersionString")
-    require(!interBrokerUsesSasl || saslEnabledMechanisms.contains(saslMechanismInterBrokerProtocol),
+    require(!interBrokerUsesSasl || saslEnabledMechanisms(interBrokerListenerName).contains(saslMechanismInterBrokerProtocol),
       s"${KafkaConfig.SaslMechanismInterBrokerProtocolProp} must be included in ${KafkaConfig.SaslEnabledMechanismsProp} when SASL is used for inter-broker communication")
     require(queuedMaxBytes <= 0 || queuedMaxBytes >= socketRequestMaxBytes,
       s"${KafkaConfig.QueuedMaxBytesProp} must be larger or equal to ${KafkaConfig.SocketRequestMaxBytesProp}")
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 747a0df..0212181 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -44,6 +44,7 @@ import org.apache.kafka.common.network._
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests.{ControlledShutdownRequest, ControlledShutdownResponse}
 import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.security.scram.ScramMechanism
 import org.apache.kafka.common.security.token.delegation.DelegationTokenCache
 import org.apache.kafka.common.security.{JaasContext, JaasUtils}
 import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time}
@@ -236,8 +237,10 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
         logManager.startup()
 
         metadataCache = new MetadataCache(config.brokerId)
-        tokenCache = new DelegationTokenCache(config.saslEnabledMechanisms)
-        credentialProvider = new CredentialProvider(config.saslEnabledMechanisms, tokenCache)
+        // Enable delegation token cache for all SCRAM mechanisms to simplify dynamic update.
+        // This keeps the cache up-to-date if new SCRAM mechanisms are enabled dynamically.
+        tokenCache = new DelegationTokenCache(ScramMechanism.mechanismNames)
+        credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, tokenCache)
 
         socketServer = new SocketServer(config, metrics, time, credentialProvider)
         socketServer.startup()
@@ -366,7 +369,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
     zkClient.getClusterId.getOrElse(zkClient.createOrGetClusterId(CoreUtils.generateUuidAsBase64))
   }
 
-  private def createBrokerInfo: BrokerInfo = {
+  private[server] def createBrokerInfo: BrokerInfo = {
     val listeners = config.advertisedListeners.map { endpoint =>
       if (endpoint.port == 0)
         endpoint.copy(port = socketServer.boundPort(endpoint.listenerName))
diff --git a/core/src/main/scala/kafka/utils/PasswordEncoder.scala b/core/src/main/scala/kafka/utils/PasswordEncoder.scala
new file mode 100644
index 0000000..ff11e24
--- /dev/null
+++ b/core/src/main/scala/kafka/utils/PasswordEncoder.scala
@@ -0,0 +1,175 @@
+/**
+ * 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 kafka.utils
+
+import java.nio.charset.StandardCharsets
+import java.security.{AlgorithmParameters, NoSuchAlgorithmException, SecureRandom}
+import java.security.spec.AlgorithmParameterSpec
+import javax.crypto.{Cipher, SecretKeyFactory}
+import javax.crypto.spec._
+
+import kafka.utils.PasswordEncoder._
+import org.apache.kafka.common.config.ConfigException
+import org.apache.kafka.common.config.types.Password
+import org.apache.kafka.common.utils.Base64
+
+import scala.collection.Map
+
+object PasswordEncoder {
+  val KeyFactoryAlgorithmProp = "keyFactoryAlgorithm"
+  val CipherAlgorithmProp = "cipherAlgorithm"
+  val InitializationVectorProp = "initializationVector"
+  val KeyLengthProp = "keyLength"
+  val SaltProp = "salt"
+  val IterationsProp = "iterations"
+  val EncyrptedPasswordProp = "encryptedPassword"
+  val PasswordLengthProp = "passwordLength"
+}
+
+/**
+  * Password encoder and decoder implementation. Encoded passwords are persisted as a CSV map
+  * containing the encoded password in base64 and along with the properties used for encryption.
+  *
+  * @param secret The secret used for encoding and decoding
+  * @param keyFactoryAlgorithm  Key factory algorithm if configured. By default, PBKDF2WithHmacSHA512 is
+  *                             used if available, PBKDF2WithHmacSHA1 otherwise.
+  * @param cipherAlgorithm Cipher algorithm used for encoding.
+  * @param keyLength Key length used for encoding. This should be valid for the specified algorithms.
+  * @param iterations Iteration count used for encoding.
+  *
+  * The provided `keyFactoryAlgorithm`, 'cipherAlgorithm`, `keyLength` and `iterations` are used for encoding passwords.
+  * The values used for encoding are stored along with the encoded password and the stored values are used for decoding.
+  *
+  */
+class PasswordEncoder(secret: Password,
+                      keyFactoryAlgorithm: Option[String],
+                      cipherAlgorithm: String,
+                      keyLength: Int,
+                      iterations: Int) extends Logging {
+
+  private val secureRandom = new SecureRandom
+  private val cipherParamsEncoder = cipherParamsInstance(cipherAlgorithm)
+
+  def encode(password: Password): String = {
+    val salt = new Array[Byte](256)
+    secureRandom.nextBytes(salt)
+    val cipher = Cipher.getInstance(cipherAlgorithm)
+    val keyFactory = secretKeyFactory(keyFactoryAlgorithm)
+    val keySpec = secretKeySpec(keyFactory, cipherAlgorithm, keyLength, salt, iterations)
+    cipher.init(Cipher.ENCRYPT_MODE, keySpec)
+    val encryptedPassword = cipher.doFinal(password.value.getBytes(StandardCharsets.UTF_8))
+    val encryptedMap = Map(
+      KeyFactoryAlgorithmProp -> keyFactory.getAlgorithm,
+      CipherAlgorithmProp -> cipherAlgorithm,
+      KeyLengthProp -> keyLength,
+      SaltProp -> base64Encode(salt),
+      IterationsProp -> iterations.toString,
+      EncyrptedPasswordProp -> base64Encode(encryptedPassword),
+      PasswordLengthProp -> password.value.length
+    ) ++ cipherParamsEncoder.toMap(cipher.getParameters)
+    encryptedMap.map { case (k, v) => s"$k:$v" }.mkString(",")
+  }
+
+  def decode(encodedPassword: String): Password = {
+    val params = CoreUtils.parseCsvMap(encodedPassword)
+    val keyFactoryAlg = params(KeyFactoryAlgorithmProp)
+    val cipherAlg = params(CipherAlgorithmProp)
+    val keyLength = params(KeyLengthProp).toInt
+    val salt = base64Decode(params(SaltProp))
+    val iterations = params(IterationsProp).toInt
+    val encryptedPassword = base64Decode(params(EncyrptedPasswordProp))
+    val passwordLengthProp = params(PasswordLengthProp).toInt
+    val cipher = Cipher.getInstance(cipherAlg)
+    val keyFactory = secretKeyFactory(Some(keyFactoryAlg))
+    val keySpec = secretKeySpec(keyFactory, cipherAlg, keyLength, salt, iterations)
+    cipher.init(Cipher.DECRYPT_MODE, keySpec, cipherParamsEncoder.toParameterSpec(params))
+    val password = try {
+      val decrypted = cipher.doFinal(encryptedPassword)
+      new String(decrypted, StandardCharsets.UTF_8)
+    } catch {
+      case e: Exception => throw new ConfigException("Password could not be decoded", e)
+    }
+    if (password.length != passwordLengthProp) // Sanity check
+      throw new ConfigException("Password could not be decoded, sanity check of length failed")
+    new Password(password)
+  }
+
+  private def secretKeyFactory(keyFactoryAlg: Option[String]): SecretKeyFactory = {
+    keyFactoryAlg match {
+      case Some(algorithm) => SecretKeyFactory.getInstance(algorithm)
+      case None =>
+        try {
+          SecretKeyFactory.getInstance("PBKDF2WithHmacSHA512")
+        } catch {
+          case _: NoSuchAlgorithmException => SecretKeyFactory.getInstance("PBKDF2WithHmacSHA1")
+        }
+    }
+  }
+
+  private def secretKeySpec(keyFactory: SecretKeyFactory,
+                            cipherAlg: String,
+                            keyLength: Int,
+                            salt: Array[Byte], iterations: Int): SecretKeySpec = {
+    val keySpec = new PBEKeySpec(secret.value.toCharArray, salt, iterations, keyLength)
+    val algorithm = if (cipherAlg.indexOf('/') > 0) cipherAlg.substring(0, cipherAlg.indexOf('/')) else cipherAlg
+    new SecretKeySpec(keyFactory.generateSecret(keySpec).getEncoded, algorithm)
+  }
+
+  private def base64Encode(bytes: Array[Byte]): String = Base64.encoder.encodeToString(bytes)
+
+  private[utils] def base64Decode(encoded: String): Array[Byte] = Base64.decoder.decode(encoded)
+
+  private def cipherParamsInstance(cipherAlgorithm: String): CipherParamsEncoder = {
+    val aesPattern = "AES/(.*)/.*".r
+    cipherAlgorithm match {
+      case aesPattern("GCM") => new GcmParamsEncoder
+      case _ => new IvParamsEncoder
+    }
+  }
+
+  private trait CipherParamsEncoder {
+    def toMap(cipher: AlgorithmParameters): Map[String, String]
+    def toParameterSpec(paramMap: Map[String, String]): AlgorithmParameterSpec
+  }
+
+  private class IvParamsEncoder extends CipherParamsEncoder {
+    def toMap(cipherParams: AlgorithmParameters): Map[String, String] = {
+      if (cipherParams != null) {
+        val ivSpec = cipherParams.getParameterSpec(classOf[IvParameterSpec])
+        Map(InitializationVectorProp -> base64Encode(ivSpec.getIV))
+      } else
+        throw new IllegalStateException("Could not determine initialization vector for cipher")
+    }
+    def toParameterSpec(paramMap: Map[String, String]): AlgorithmParameterSpec = {
+      new IvParameterSpec(base64Decode(paramMap(InitializationVectorProp)))
+    }
+  }
+
+  private class GcmParamsEncoder extends CipherParamsEncoder {
+    def toMap(cipherParams: AlgorithmParameters): Map[String, String] = {
+      if (cipherParams != null) {
+        val spec = cipherParams.getParameterSpec(classOf[GCMParameterSpec])
+        Map(InitializationVectorProp -> base64Encode(spec.getIV),
+            "authenticationTagLength" -> spec.getTLen.toString)
+      } else
+        throw new IllegalStateException("Could not determine initialization vector for cipher")
+    }
+    def toParameterSpec(paramMap: Map[String, String]): AlgorithmParameterSpec = {
+      new GCMParameterSpec(paramMap("authenticationTagLength").toInt, base64Decode(paramMap(InitializationVectorProp)))
+    }
+  }
+}
diff --git a/core/src/main/scala/kafka/zk/KafkaZkClient.scala b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
index d683a8d..b545455 100644
--- a/core/src/main/scala/kafka/zk/KafkaZkClient.scala
+++ b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
@@ -82,6 +82,13 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
     info(s"Registered broker ${brokerInfo.broker.id} at path $path with addresses: ${brokerInfo.broker.endPoints}")
   }
 
+  def updateBrokerInfoInZk(brokerInfo: BrokerInfo): Unit = {
+    val brokerIdPath = brokerInfo.path
+    val setDataRequest = SetDataRequest(brokerIdPath, brokerInfo.toJsonBytes, ZkVersion.NoVersion)
+    retryRequestUntilConnected(setDataRequest)
+    info("Updated broker %d at path %s with addresses: %s".format(brokerInfo.broker.id, brokerIdPath, brokerInfo.broker.endPoints))
+  }
+
   /**
    * Gets topic partition states for the given partitions.
    * @param partitions the partitions for which we want ot get states.
diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index 6f45bca..867e03d 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -430,13 +430,13 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
     assertEquals(KafkaConfig.ListenerSecurityProtocolMapProp, listenerSecurityProtocolMap.name)
     assertFalse(listenerSecurityProtocolMap.isDefault)
     assertFalse(listenerSecurityProtocolMap.isSensitive)
-    assertTrue(listenerSecurityProtocolMap.isReadOnly)
+    assertFalse(listenerSecurityProtocolMap.isReadOnly)
     val truststorePassword = configs.get(brokerResource1).get(KafkaConfig.SslTruststorePasswordProp)
     assertEquals(KafkaConfig.SslTruststorePasswordProp, truststorePassword.name)
     assertNull(truststorePassword.value)
     assertFalse(truststorePassword.isDefault)
     assertTrue(truststorePassword.isSensitive)
-    assertTrue(truststorePassword.isReadOnly)
+    assertFalse(truststorePassword.isReadOnly)
     val compressionType = configs.get(brokerResource1).get(KafkaConfig.CompressionTypeProp)
     assertEquals(servers(1).config.compressionType.toString, compressionType.value)
     assertEquals(KafkaConfig.CompressionTypeProp, compressionType.name)
diff --git a/core/src/test/scala/integration/kafka/api/SaslSetup.scala b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
index f459252..273b247 100644
--- a/core/src/test/scala/integration/kafka/api/SaslSetup.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
@@ -59,12 +59,7 @@ trait SaslSetup {
       case _ => false
     })
     if (hasKerberos) {
-      val (serverKeytabFile, clientKeytabFile) = maybeCreateEmptyKeytabFiles()
-      kdc = new MiniKdc(kdcConf, workDir)
-      kdc.start()
-      kdc.createPrincipal(serverKeytabFile, JaasTestUtils.KafkaServerPrincipalUnqualifiedName + "/localhost")
-      kdc.createPrincipal(clientKeytabFile,
-        JaasTestUtils.KafkaClientPrincipalUnqualifiedName, JaasTestUtils.KafkaClientPrincipalUnqualifiedName2)
+      initializeKerberos()
     }
     writeJaasConfigurationToFile(jaasSections)
     val hasZk = jaasSections.exists(_.modules.exists {
@@ -75,6 +70,15 @@ trait SaslSetup {
       System.setProperty("zookeeper.authProvider.1", "org.apache.zookeeper.server.auth.SASLAuthenticationProvider")
   }
 
+  protected def initializeKerberos(): Unit = {
+    val (serverKeytabFile, clientKeytabFile) = maybeCreateEmptyKeytabFiles()
+    kdc = new MiniKdc(kdcConf, workDir)
+    kdc.start()
+    kdc.createPrincipal(serverKeytabFile, JaasTestUtils.KafkaServerPrincipalUnqualifiedName + "/localhost")
+    kdc.createPrincipal(clientKeytabFile,
+      JaasTestUtils.KafkaClientPrincipalUnqualifiedName, JaasTestUtils.KafkaClientPrincipalUnqualifiedName2)
+  }
+
   /** Return a tuple with the path to the server keytab file and client keytab file */
   protected def maybeCreateEmptyKeytabFiles(): (File, File) = {
     if (serverKeytabFile.isEmpty)
diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
index 1224274..b7f0ae8 100644
--- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
@@ -18,28 +18,29 @@
 
 package kafka.server
 
-import java.io.{Closeable, File, FileOutputStream, FileWriter}
+import java.io.{Closeable, File, FileWriter}
 import java.nio.file.{Files, StandardCopyOption}
 import java.lang.management.ManagementFactory
 import java.util
 import java.util.{Collections, Properties}
-import java.util.concurrent.{ConcurrentLinkedQueue, ExecutionException, TimeUnit}
+import java.util.concurrent._
 import javax.management.ObjectName
 
 import kafka.admin.ConfigCommand
-import kafka.api.SaslSetup
-import kafka.log.LogConfig
+import kafka.api.{KafkaSasl, SaslSetup}
 import kafka.coordinator.group.OffsetConfig
+import kafka.log.LogConfig
 import kafka.message.ProducerCompressionCodec
-import kafka.utils.{ShutdownableThread, TestUtils}
+import kafka.utils._
 import kafka.utils.Implicits._
 import kafka.zk.{ConfigEntityChangeNotificationZNode, ZooKeeperTestHarness}
+import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.admin.ConfigEntry.{ConfigSource, ConfigSynonym}
 import org.apache.kafka.clients.admin._
 import org.apache.kafka.clients.consumer.{ConsumerRecord, ConsumerRecords, KafkaConsumer}
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.{ClusterResource, ClusterResourceListener, Reconfigurable, TopicPartition}
-import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.{ConfigException, ConfigResource, SslConfigs}
 import org.apache.kafka.common.config.SslConfigs._
 import org.apache.kafka.common.config.types.Password
 import org.apache.kafka.common.errors.{AuthenticationException, InvalidRequestException}
@@ -55,6 +56,7 @@ import org.junit.{After, Before, Test}
 import scala.collection._
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.JavaConverters._
+import scala.collection.Seq
 
 object DynamicBrokerReconfigurationTest {
   val SecureInternal = "INTERNAL"
@@ -65,12 +67,13 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
 
   import DynamicBrokerReconfigurationTest._
 
-  private var servers = new ArrayBuffer[KafkaServer]
+  private val servers = new ArrayBuffer[KafkaServer]
   private val numServers = 3
   private val producers = new ArrayBuffer[KafkaProducer[String, String]]
   private val consumers = new ArrayBuffer[KafkaConsumer[String, String]]
   private val adminClients = new ArrayBuffer[AdminClient]()
   private val clientThreads = new ArrayBuffer[ShutdownableThread]()
+  private val executors = new ArrayBuffer[ExecutorService]
   private val topic = "testtopic"
 
   private val kafkaClientSaslMechanism = "PLAIN"
@@ -95,10 +98,13 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
       props.put(KafkaConfig.ListenersProp, s"$SecureInternal://localhost:0, $SecureExternal://localhost:0")
       props.put(KafkaConfig.ListenerSecurityProtocolMapProp, s"$SecureInternal:SSL, $SecureExternal:SASL_SSL")
       props.put(KafkaConfig.InterBrokerListenerNameProp, SecureInternal)
+      props.put(KafkaConfig.SaslMechanismInterBrokerProtocolProp, "PLAIN")
       props.put(KafkaConfig.ZkEnableSecureAclsProp, "true")
       props.put(KafkaConfig.SaslEnabledMechanismsProp, kafkaServerSaslMechanisms.mkString(","))
       props.put(KafkaConfig.LogSegmentBytesProp, "2000")
       props.put(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp, "10000000")
+      props.put(KafkaConfig.PasswordEncoderSecretProp, "dynamic-config-secret")
+      props.put(KafkaConfig.PasswordEncoderOldSecretProp, "old-dynamic-config-secret")
 
       props ++= sslProperties1
       addKeystoreWithListenerPrefix(sslProperties1, props, SecureInternal)
@@ -127,8 +133,9 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     clientThreads.foreach(_.interrupt())
     clientThreads.foreach(_.initiateShutdown())
     clientThreads.foreach(_.join(5 * 1000))
-    producers.foreach(_.close())
-    consumers.foreach(_.close())
+    executors.foreach(_.shutdownNow())
+    producers.foreach(_.close(0, TimeUnit.MILLISECONDS))
+    consumers.foreach(_.close(0, TimeUnit.MILLISECONDS))
     adminClients.foreach(_.close())
     TestUtils.shutdownServers(servers)
     super.tearDown()
@@ -514,13 +521,217 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     stopAndVerifyProduceConsume(producerThread, consumerThread)
   }
 
+  @Test
+  def testAdvertisedListenerUpdate(): Unit = {
+    val adminClient = adminClients.head
+    val externalAdminClient = createAdminClient(SecurityProtocol.SASL_SSL, SecureExternal)
+
+    // Ensure connections are made to brokers before external listener is made inaccessible
+    describeConfig(externalAdminClient)
+
+    // Update broker keystore for external listener to use invalid listener address
+    // any address other than localhost is sufficient to fail (either connection or host name verification failure)
+    val invalidHost = "192.168.0.1"
+    alterAdvertisedListener(adminClient, externalAdminClient, "localhost", invalidHost)
+
+    // Verify that producer connections fail since advertised listener is invalid
+    val bootstrap = bootstrapServers.replaceAll(invalidHost, "localhost") // allow bootstrap connection to succeed
+    val producer1 = createProducer(trustStoreFile1, retries = 0, bootstrap = bootstrap)
+
+    val sendFuture = verifyConnectionFailure(producer1)
+
+    alterAdvertisedListener(adminClient, externalAdminClient, invalidHost, "localhost")
+
+    // Verify that produce/consume work now
+    val producer = createProducer(trustStoreFile1, retries = 0)
+    val consumer = createConsumer("group2", trustStoreFile1, topic)
+    verifyProduceConsume(producer, consumer, 10, topic)
+
+    // Verify updating inter-broker listener
+    val props = new Properties
+    props.put(KafkaConfig.InterBrokerListenerNameProp, SecureExternal)
+    try {
+      reconfigureServers(props, perBrokerConfig = true, (KafkaConfig.InterBrokerListenerNameProp, SecureExternal))
+      fail("Inter-broker listener cannot be dynamically updated")
+    } catch {
+      case e: ExecutionException =>
+        assertTrue(s"Unexpected exception ${e.getCause}", e.getCause.isInstanceOf[InvalidRequestException])
+        servers.foreach(server => assertEquals(SecureInternal, server.config.interBrokerListenerName.value))
+    }
+
+    // Verify that the other send did not complete
+    verifyTimeout(sendFuture)
+  }
+
+  @Test
+  def testAddRemoveSslListener(): Unit = {
+    verifyAddListener("SSL", SecurityProtocol.SSL, Seq.empty)
+
+    // Restart servers and check secret rotation
+    servers.foreach(_.shutdown())
+    servers.foreach(_.awaitShutdown())
+    adminClients.foreach(_.close())
+    adminClients.clear()
+
+    // All passwords are currently encoded with password.encoder.secret. Encode with password.encoder.old.secret
+    // and update ZK. When each server is started, it should decode using password.encoder.old.secret and update
+    // ZK with newly encoded values using password.encoder.secret.
+    servers.foreach { server =>
+      val props = adminZkClient.fetchEntityConfig(ConfigType.Broker, server.config.brokerId.toString)
+      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 passwordDecoder = new PasswordEncoder(secret,
+        config.passwordEncoderKeyFactoryAlgorithm,
+        config.passwordEncoderCipherAlgorithm,
+        config.passwordEncoderKeyLength,
+        config.passwordEncoderIterations)
+      val passwordEncoder = new PasswordEncoder(oldSecret,
+        config.passwordEncoderKeyFactoryAlgorithm,
+        config.passwordEncoderCipherAlgorithm,
+        config.passwordEncoderKeyLength,
+        config.passwordEncoderIterations)
+      passwordConfigs.foreach { case (name, value) =>
+          val decoded = passwordDecoder.decode(value).value
+          props.put(name, passwordEncoder.encode(new Password(decoded)))
+      }
+      val brokerId = server.config.brokerId
+      adminZkClient.changeBrokerConfig(Seq(brokerId), props)
+      val updatedProps = adminZkClient.fetchEntityConfig(ConfigType.Broker, brokerId.toString)
+      passwordConfigs.foreach { case (name, value) => assertNotEquals(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)) }
+      }
+    }
+
+    verifyListener(SecurityProtocol.SSL, None)
+    createAdminClient(SecurityProtocol.SSL, SecureInternal)
+    verifyRemoveListener("SSL", SecurityProtocol.SSL, Seq.empty)
+  }
+
+  @Test
+  def testAddRemoveSaslListeners(): Unit = {
+    createScramCredentials(zkConnect, JaasTestUtils.KafkaScramUser, JaasTestUtils.KafkaScramPassword)
+    createScramCredentials(zkConnect, JaasTestUtils.KafkaScramAdmin, JaasTestUtils.KafkaScramAdminPassword)
+    initializeKerberos()
+
+    //verifyAddListener("SASL_SSL", SecurityProtocol.SASL_SSL, Seq("SCRAM-SHA-512", "SCRAM-SHA-256", "PLAIN"))
+    verifyAddListener("SASL_PLAINTEXT", SecurityProtocol.SASL_PLAINTEXT, Seq("GSSAPI"))
+    //verifyRemoveListener("SASL_SSL", SecurityProtocol.SASL_SSL, Seq("SCRAM-SHA-512", "SCRAM-SHA-256", "PLAIN"))
+    verifyRemoveListener("SASL_PLAINTEXT", SecurityProtocol.SASL_PLAINTEXT, Seq("GSSAPI"))
+  }
+
+  private def verifyAddListener(listenerName: String, securityProtocol: SecurityProtocol,
+                                saslMechanisms: Seq[String]): Unit = {
+    val config = servers.head.config
+    val existingListenerCount = config.listeners.size
+    val listeners = config.listeners
+      .map(e => s"${e.listenerName.value}://${e.host}:${e.port}")
+      .mkString(",") + s",$listenerName://localhost:0"
+    val listenerMap = config.listenerSecurityProtocolMap
+      .map { case (name, protocol) => s"${name.value}:${protocol.name}" }
+      .mkString(",") + s",$listenerName:${securityProtocol.name}"
+
+    val props =  adminZkClient.fetchEntityConfig(ConfigType.Broker, config.brokerId.toString)
+    props.put(KafkaConfig.ListenersProp, listeners)
+    props.put(KafkaConfig.ListenerSecurityProtocolMapProp, listenerMap)
+    securityProtocol match {
+      case SecurityProtocol.SSL =>
+        addListenerPropsSsl(listenerName, props)
+      case SecurityProtocol.SASL_PLAINTEXT =>
+        addListenerPropsSasl(listenerName, saslMechanisms, props)
+      case SecurityProtocol.SASL_SSL =>
+        addListenerPropsSasl(listenerName, saslMechanisms, props)
+        addListenerPropsSsl(listenerName, props)
+      case SecurityProtocol.PLAINTEXT => // no additional props
+    }
+
+    alterConfigs(adminClients.head, props, perBrokerConfig = true).all.get
+
+    TestUtils.waitUntilTrue(() => servers.forall(server => server.config.listeners.size == existingListenerCount + 1),
+      "Listener config not updated")
+    TestUtils.waitUntilTrue(() => servers.forall(server => {
+      try {
+        server.socketServer.boundPort(new ListenerName(listenerName)) > 0
+      } catch {
+        case _: Exception => false
+      }
+    }), "Listener not created")
+
+    if (saslMechanisms.nonEmpty)
+      saslMechanisms.foreach(mechanism => verifyListener(securityProtocol, Some(mechanism)))
+    else
+      verifyListener(securityProtocol, None)
+  }
+
+  private def verifyRemoveListener(listenerName: String, securityProtocol: SecurityProtocol,
+                                   saslMechanisms: Seq[String]): Unit = {
+    val saslMechanism = if (saslMechanisms.isEmpty) "" else saslMechanisms.head
+    val producer1 = createProducer(listenerName, securityProtocol, saslMechanism)
+    val consumer1 = createConsumer(listenerName, securityProtocol, saslMechanism,
+      s"remove-listener-group-$securityProtocol")
+    verifyProduceConsume(producer1, consumer1, numRecords = 10, topic)
+    // send another message to check consumer later
+    producer1.send(new ProducerRecord(topic, "key", "value")).get(100, TimeUnit.MILLISECONDS)
+
+    val config = servers.head.config
+    val existingListenerCount = config.listeners.size
+    val listeners = config.listeners
+      .filter(e => e.listenerName.value != securityProtocol.name)
+      .map(e => s"${e.listenerName.value}://${e.host}:${e.port}")
+      .mkString(",")
+    val listenerMap = config.listenerSecurityProtocolMap
+      .filterKeys(listenerName => listenerName.value != securityProtocol.name)
+      .map { case (listenerName, protocol) => s"${listenerName.value}:${protocol.name}" }
+      .mkString(",")
+
+    val props = adminZkClient.fetchEntityConfig(ConfigType.Broker, config.brokerId.toString)
+    val listenerProps = props.asScala.keySet.filter(_.startsWith(new ListenerName(listenerName).configPrefix))
+    listenerProps.foreach(props.remove)
+    props.put(KafkaConfig.ListenersProp, listeners)
+    props.put(KafkaConfig.ListenerSecurityProtocolMapProp, listenerMap)
+    alterConfigs(adminClients.head, props, perBrokerConfig = true).all.get
+
+    TestUtils.waitUntilTrue(() => servers.forall(server => server.config.listeners.size == existingListenerCount - 1),
+      "Listeners not updated")
+
+    // Test that connections using deleted listener don't work
+    val producerFuture = verifyConnectionFailure(producer1)
+    val consumerFuture = verifyConnectionFailure(consumer1)
+
+    // Test that other listeners still work
+    val producer2 = createProducer(trustStoreFile1, retries = 0)
+    val consumer2 = createConsumer(s"remove-listener-group2-$securityProtocol", trustStoreFile1, topic, autoOffsetReset = "latest")
+    verifyProduceConsume(producer2, consumer2, numRecords = 10, topic)
+
+    // Verify that producer/consumer using old listener don't work
+    verifyTimeout(producerFuture)
+    verifyTimeout(consumerFuture)
+  }
+
+  private def verifyListener(securityProtocol: SecurityProtocol, saslMechanism: Option[String]): Unit = {
+    val mechanism = saslMechanism.getOrElse("")
+    val producer = createProducer(securityProtocol.name, securityProtocol, mechanism)
+    val consumer = createConsumer(securityProtocol.name, securityProtocol, mechanism,
+      s"add-listener-group-$securityProtocol-$mechanism")
+    verifyProduceConsume(producer, consumer, numRecords = 10, topic)
+  }
+
+  private def bootstrapServers: String = TestUtils.bootstrapServers(servers, new ListenerName(SecureExternal))
+
   private def createProducer(trustStore: File, retries: Int,
-                             clientId: String = "test-producer"): KafkaProducer[String, String] = {
-    val bootstrapServers = TestUtils.bootstrapServers(servers, new ListenerName(SecureExternal))
+                             clientId: String = "test-producer",
+                             bootstrap: String = bootstrapServers,
+                             securityProtocol: SecurityProtocol = SecurityProtocol.SASL_SSL): KafkaProducer[String, String] = {
     val propsOverride = new Properties
     propsOverride.put(ProducerConfig.CLIENT_ID_CONFIG, clientId)
+    propsOverride.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "HTTPS")
     val producer = TestUtils.createNewProducer(
-      bootstrapServers,
+      bootstrap,
       acks = -1,
       retries = retries,
       securityProtocol = SecurityProtocol.SASL_SSL,
@@ -533,17 +744,72 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     producer
   }
 
-  private def createConsumer(groupId: String, trustStore: File, topic: String = topic):KafkaConsumer[String, String] = {
-    val bootstrapServers = TestUtils.bootstrapServers(servers, new ListenerName(SecureExternal))
+  private def createConsumer(groupId: String, trustStore: File,
+                             topic: String = topic,
+                             bootstrap: String = bootstrapServers,
+                             securityProtocol: SecurityProtocol = SecurityProtocol.SASL_SSL,
+                             autoOffsetReset: String = "earliest"):KafkaConsumer[String, String] = {
+    val propsOverride = new Properties
+    propsOverride.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "HTTPS")
     val consumer = TestUtils.createNewConsumer(
-      bootstrapServers,
+      bootstrap,
       groupId,
-      securityProtocol = SecurityProtocol.SASL_SSL,
+      autoOffsetReset = autoOffsetReset,
+      securityProtocol = securityProtocol,
       trustStoreFile = Some(trustStore),
       saslProperties = Some(clientSaslProps),
       keyDeserializer = new StringDeserializer,
       valueDeserializer = new StringDeserializer)
     consumer.subscribe(Collections.singleton(topic))
+    if (autoOffsetReset == "latest") {
+      do {
+        consumer.poll(1)
+      } while (consumer.assignment.isEmpty)
+    }
+    consumers += consumer
+    consumer
+  }
+
+  private def clientProps(securityProtocol: SecurityProtocol, saslMechanism: String): Properties = {
+    val props = new Properties
+    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol.name)
+    props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "HTTPS")
+    val saslProps = if (securityProtocol == SecurityProtocol.SASL_PLAINTEXT || securityProtocol == SecurityProtocol.SASL_SSL) {
+      Some(kafkaClientSaslProperties(saslMechanism, dynamicJaasConfig = true))
+    } else
+      None
+    val securityProps = TestUtils.securityConfigs(Mode.CLIENT, securityProtocol,
+      Some(trustStoreFile1), "client", TestUtils.SslCertificateCn, saslProps)
+    props ++= securityProps
+    props
+  }
+
+  private def createProducer(listenerName: String, securityProtocol: SecurityProtocol,
+                             saslMechanism: String): KafkaProducer[String, String] = {
+    val bootstrapServers =  TestUtils.bootstrapServers(servers, new ListenerName(listenerName))
+    val producer = TestUtils.createNewProducer(bootstrapServers,
+      acks = -1, retries = 0,
+      securityProtocol = securityProtocol,
+      keySerializer = new StringSerializer,
+      valueSerializer = new StringSerializer,
+      props = Some(clientProps(securityProtocol, saslMechanism)))
+    producers += producer
+    producer
+  }
+
+  private def createConsumer(listenerName: String, securityProtocol: SecurityProtocol,
+                             saslMechanism: String, group: String): KafkaConsumer[String, String] = {
+    val bootstrapServers =  TestUtils.bootstrapServers(servers, new ListenerName(listenerName))
+    val consumer = TestUtils.createNewConsumer(bootstrapServers, group,
+      autoOffsetReset = "latest",
+      securityProtocol = securityProtocol,
+      keyDeserializer = new StringDeserializer,
+      valueDeserializer = new StringDeserializer,
+      props = Some(clientProps(securityProtocol, saslMechanism)))
+    consumer.subscribe(Collections.singleton(topic))
+    do {
+      consumer.poll(1)
+    } while (consumer.assignment.isEmpty)
     consumers += consumer
     consumer
   }
@@ -552,6 +818,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     val config = new util.HashMap[String, Object]
     val bootstrapServers = TestUtils.bootstrapServers(servers, new ListenerName(listenerName))
     config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
+    config.put(AdminClientConfig.METADATA_MAX_AGE_CONFIG, "10")
     val securityProps: util.Map[Object, Object] =
       TestUtils.adminClientSecurityConfigs(securityProtocol, Some(trustStoreFile1), Some(clientSaslProps))
     securityProps.asScala.foreach { case (key, value) => config.put(key.asInstanceOf[String], value) }
@@ -563,7 +830,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
   private def verifyProduceConsume(producer: KafkaProducer[String, String],
                                    consumer: KafkaConsumer[String, String],
                                    numRecords: Int,
-                                   topic: String = topic): Unit = {
+                                   topic: String): Unit = {
     val producerRecords = (1 to numRecords).map(i => new ProducerRecord(topic, s"key$i", s"value$i"))
     producerRecords.map(producer.send).map(_.get(10, TimeUnit.SECONDS))
 
@@ -639,6 +906,39 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     waitForConfig(s"$configPrefix$SSL_KEYSTORE_LOCATION_CONFIG", props.getProperty(SSL_KEYSTORE_LOCATION_CONFIG))
   }
 
+  private def serverEndpoints(adminClient: AdminClient): String = {
+    val nodes = adminClient.describeCluster().nodes().get
+    nodes.asScala.map { node =>
+      s"${node.host}:${node.port}"
+    }.mkString(",")
+  }
+
+  private def alterAdvertisedListener(adminClient: AdminClient, externalAdminClient: AdminClient, oldHost: String, newHost: String): Unit = {
+    val configs = servers.map { server =>
+      val resource = new ConfigResource(ConfigResource.Type.BROKER, server.config.brokerId.toString)
+      val newListeners = server.config.advertisedListeners.map { e =>
+        if (e.listenerName.value == SecureExternal)
+          s"${e.listenerName.value}://$newHost:${server.boundPort(e.listenerName)}"
+        else
+          s"${e.listenerName.value}://${e.host}:${server.boundPort(e.listenerName)}"
+      }.mkString(",")
+      val configEntry = new ConfigEntry(KafkaConfig.AdvertisedListenersProp, newListeners)
+      (resource, new Config(Collections.singleton(configEntry)))
+    }.toMap.asJava
+    adminClient.alterConfigs(configs).all.get
+    servers.foreach { server =>
+      TestUtils.retry(10000) {
+        val externalListener = server.config.advertisedListeners.find(_.listenerName.value == SecureExternal)
+          .getOrElse(throw new IllegalStateException("External listener not found"))
+        assertTrue("Config not updated", externalListener.host == newHost)
+      }
+    }
+    val (endpoints, altered) = TestUtils.computeUntilTrue(serverEndpoints(externalAdminClient)) { endpoints =>
+      !endpoints.contains(oldHost)
+    }
+    assertTrue(s"Advertised listener update not propagated by controller: $endpoints", altered)
+  }
+
   private def alterConfigs(adminClient: AdminClient, props: Properties, perBrokerConfig: Boolean): AlterConfigsResult = {
     val configEntries = props.asScala.map { case (k, v) => new ConfigEntry(k, v) }.toList.asJava
     val newConfig = new Config(configEntries)
@@ -701,12 +1001,6 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     adminZkClient.changeBrokerConfig(brokers, keystoreProps)
   }
 
-  private def waitForKeystore(sslProperties: Properties, maxWaitMs: Long = 10000): Unit = {
-    waitForConfig(new ListenerName(SecureExternal).configPrefix + SSL_KEYSTORE_LOCATION_CONFIG,
-      sslProperties.getProperty(SSL_KEYSTORE_LOCATION_CONFIG), maxWaitMs)
-
-  }
-
   private def waitForConfig(propName: String, propValue: String, maxWaitMs: Long = 10000): Unit = {
     servers.foreach { server => waitForConfigOnServer(server, propName, propValue, maxWaitMs) }
   }
@@ -774,6 +1068,64 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     }
   }
 
+  private def verifyConnectionFailure(producer: KafkaProducer[String, String]): Future[_] = {
+    val executor = Executors.newSingleThreadExecutor
+    executors += executor
+    val future = executor.submit(new Runnable() {
+      def run() {
+        producer.send(new ProducerRecord(topic, "key", "value")).get
+      }
+    })
+    verifyTimeout(future)
+    future
+  }
+
+  private def verifyConnectionFailure(consumer: KafkaConsumer[String, String]): Future[_] = {
+    val executor = Executors.newSingleThreadExecutor
+    executors += executor
+    val future = executor.submit(new Runnable() {
+      def run() {
+        assertEquals(0, consumer.poll(100).count)
+      }
+    })
+    verifyTimeout(future)
+    future
+  }
+
+  private def verifyTimeout(future: Future[_]): Unit = {
+    try {
+      future.get(100, TimeUnit.MILLISECONDS)
+      fail("Operation should not have completed")
+    } catch {
+      case _: TimeoutException => // expected exception
+    }
+  }
+
+  private def addListenerPropsSsl(listenerName: String, props: Properties): Unit = {
+    val prefix = new ListenerName(listenerName).configPrefix
+    sslProperties1.keySet.asScala.foreach { name =>
+      val value = sslProperties1.get(name)
+      val valueStr = value match {
+        case password: Password => password.value
+        case list: util.List[_] => list.asScala.map(_.toString).mkString(",")
+        case _ => value.toString
+      }
+      props.put(s"$prefix$name", valueStr)
+    }
+  }
+
+  private def addListenerPropsSasl(listener: String, mechanisms: Seq[String], props: Properties): Unit = {
+    val listenerName = new ListenerName(listener)
+    val prefix = listenerName.configPrefix
+    props.put(prefix + KafkaConfig.SaslEnabledMechanismsProp, mechanisms.mkString(","))
+    props.put(prefix + KafkaConfig.SaslKerberosServiceNameProp, "kafka")
+    mechanisms.foreach { mechanism =>
+      val jaasSection = jaasSections(Seq(mechanism), None, KafkaSasl, "").head
+      val jaasConfig = jaasSection.modules.head.toString
+      props.put(listenerName.saslMechanismConfigPrefix(mechanism) + KafkaConfig.SaslJaasConfigProp, jaasConfig)
+    }
+  }
+
   private class ProducerThread(clientId: String, retries: Int) extends ShutdownableThread(clientId, isInterruptible = false) {
     private val producer = createProducer(trustStoreFile1, retries, clientId)
     @volatile var sent = 0
@@ -877,8 +1229,10 @@ class TestMetricsReporter extends MetricsReporter with Reconfigurable with Close
     Set(PollingIntervalProp).asJava
   }
 
-  override def validateReconfiguration(configs: util.Map[String, _]): Boolean = {
-    configs.get(PollingIntervalProp).toString.toInt > 0
+  override def validateReconfiguration(configs: util.Map[String, _]): Unit = {
+    val pollingInterval = configs.get(PollingIntervalProp).toString
+    if (configs.get(PollingIntervalProp).toString.toInt <= 0)
+      throw new ConfigException(s"Invalid polling interval $pollingInterval")
   }
 
   override def reconfigure(configs: util.Map[String, _]): Unit = {
diff --git a/core/src/test/scala/integration/kafka/server/MultipleListenersWithAdditionalJaasContextTest.scala b/core/src/test/scala/integration/kafka/server/MultipleListenersWithAdditionalJaasContextTest.scala
index 666ba4b..f2c6753 100644
--- a/core/src/test/scala/integration/kafka/server/MultipleListenersWithAdditionalJaasContextTest.scala
+++ b/core/src/test/scala/integration/kafka/server/MultipleListenersWithAdditionalJaasContextTest.scala
@@ -21,25 +21,23 @@ import java.util.Properties
 
 import kafka.utils.JaasTestUtils
 import kafka.utils.JaasTestUtils.JaasSection
-import org.apache.kafka.common.network.ListenerName
 
 class MultipleListenersWithAdditionalJaasContextTest extends MultipleListenersWithSameSecurityProtocolBaseTest {
 
   import MultipleListenersWithSameSecurityProtocolBaseTest._
 
-  override def saslProperties(listenerName: ListenerName): Properties = {
-    listenerName.value match {
-      case SecureInternal => kafkaClientSaslProperties(Plain, dynamicJaasConfig = true)
-      case SecureExternal => kafkaClientSaslProperties(GssApi, dynamicJaasConfig = true)
-      case _ => throw new IllegalArgumentException(s"Unexpected listener name $listenerName")
-    }
+  override def staticJaasSections: Seq[JaasSection] = {
+    val (serverKeytabFile, _) = maybeCreateEmptyKeytabFiles()
+    JaasTestUtils.zkSections :+
+      JaasTestUtils.kafkaServerSection("secure_external.KafkaServer", kafkaServerSaslMechanisms(SecureExternal), Some(serverKeytabFile))
   }
 
-  override def jaasSections: Seq[JaasSection] = {
-    val (serverKeytabFile, _) = maybeCreateEmptyKeytabFiles()
-    JaasTestUtils.zkSections ++ Seq(
-      JaasTestUtils.kafkaServerSection("secure_external.KafkaServer", Seq(GssApi), Some(serverKeytabFile)),
-      JaasTestUtils.kafkaServerSection("secure_internal.KafkaServer", Seq(Plain), None)
-    )
+  override protected def dynamicJaasSections: Properties = {
+    val props = new Properties
+    kafkaServerSaslMechanisms(SecureInternal).foreach { mechanism =>
+      addDynamicJaasSection(props, SecureInternal, mechanism,
+        JaasTestUtils.kafkaServerSection("secure_internal.KafkaServer", Seq(mechanism), None))
+    }
+    props
   }
 }
diff --git a/core/src/test/scala/integration/kafka/server/MultipleListenersWithDefaultJaasContextTest.scala b/core/src/test/scala/integration/kafka/server/MultipleListenersWithDefaultJaasContextTest.scala
index f3e1b8b..10df84e 100644
--- a/core/src/test/scala/integration/kafka/server/MultipleListenersWithDefaultJaasContextTest.scala
+++ b/core/src/test/scala/integration/kafka/server/MultipleListenersWithDefaultJaasContextTest.scala
@@ -26,12 +26,9 @@ import org.apache.kafka.common.network.ListenerName
 
 class MultipleListenersWithDefaultJaasContextTest extends MultipleListenersWithSameSecurityProtocolBaseTest {
 
-  import MultipleListenersWithSameSecurityProtocolBaseTest._
+  override def staticJaasSections: Seq[JaasSection] =
+    jaasSections(kafkaServerSaslMechanisms.values.flatMap(identity).toSeq, Some(kafkaClientSaslMechanism), Both)
 
-  override def saslProperties(listenerName: ListenerName): Properties =
-    kafkaClientSaslProperties(Plain, dynamicJaasConfig = true)
-
-  override def jaasSections: Seq[JaasSection] =
-    jaasSections(kafkaServerSaslMechanisms, Some(kafkaClientSaslMechanism), Both)
+  override protected def dynamicJaasSections: Properties = new Properties
 
 }
diff --git a/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala b/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala
index da8437e..50ebaa4 100644
--- a/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala
+++ b/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala
@@ -19,13 +19,13 @@
 package kafka.server
 
 import java.io.File
-import java.util.{Collections, Properties}
+import java.util.{Collections, Objects, Properties}
 import java.util.concurrent.TimeUnit
 
 import kafka.api.SaslSetup
 import kafka.coordinator.group.OffsetConfig
 import kafka.utils.JaasTestUtils.JaasSection
-import kafka.utils.TestUtils
+import kafka.utils.{JaasTestUtils, TestUtils}
 import kafka.utils.Implicits._
 import kafka.zk.ZooKeeperTestHarness
 import org.apache.kafka.clients.consumer.{ConsumerRecord, KafkaConsumer}
@@ -55,18 +55,20 @@ abstract class MultipleListenersWithSameSecurityProtocolBaseTest extends ZooKeep
 
   private val trustStoreFile = File.createTempFile("truststore", ".jks")
   private val servers = new ArrayBuffer[KafkaServer]
-  private val producers = mutable.Map[ListenerName, KafkaProducer[Array[Byte], Array[Byte]]]()
-  private val consumers = mutable.Map[ListenerName, KafkaConsumer[Array[Byte], Array[Byte]]]()
+  private val producers = mutable.Map[ClientMetadata, KafkaProducer[Array[Byte], Array[Byte]]]()
+  private val consumers = mutable.Map[ClientMetadata, KafkaConsumer[Array[Byte], Array[Byte]]]()
 
   protected val kafkaClientSaslMechanism = Plain
-  protected val kafkaServerSaslMechanisms = List(GssApi, Plain)
+  protected val kafkaServerSaslMechanisms = Map(
+    SecureExternal -> Seq("SCRAM-SHA-256", GssApi),
+    SecureInternal -> Seq(Plain, "SCRAM-SHA-512"))
 
-  protected def saslProperties(listenerName: ListenerName): Properties
-  protected def jaasSections: Seq[JaasSection]
+  protected def staticJaasSections: Seq[JaasSection]
+  protected def dynamicJaasSections: Properties
 
   @Before
   override def setUp(): Unit = {
-    startSasl(jaasSections)
+    startSasl(staticJaasSections)
     super.setUp()
     // 2 brokers so that we can test that the data propagates correctly via UpdateMetadadaRequest
     val numServers = 2
@@ -82,8 +84,12 @@ abstract class MultipleListenersWithSameSecurityProtocolBaseTest extends ZooKeep
       props.put(KafkaConfig.InterBrokerListenerNameProp, Internal)
       props.put(KafkaConfig.ZkEnableSecureAclsProp, "true")
       props.put(KafkaConfig.SaslMechanismInterBrokerProtocolProp, kafkaClientSaslMechanism)
-      props.put(KafkaConfig.SaslEnabledMechanismsProp, kafkaServerSaslMechanisms.mkString(","))
+      props.put(s"${new ListenerName(SecureInternal).configPrefix}${KafkaConfig.SaslEnabledMechanismsProp}",
+        kafkaServerSaslMechanisms(SecureInternal).mkString(","))
+      props.put(s"${new ListenerName(SecureExternal).configPrefix}${KafkaConfig.SaslEnabledMechanismsProp}",
+        kafkaServerSaslMechanisms(SecureExternal).mkString(","))
       props.put(KafkaConfig.SaslKerberosServiceNameProp, "kafka")
+      props ++= dynamicJaasSections
 
       props ++= TestUtils.sslConfigs(Mode.SERVER, false, Some(trustStoreFile), s"server$brokerId")
 
@@ -107,26 +113,37 @@ abstract class MultipleListenersWithSameSecurityProtocolBaseTest extends ZooKeep
     TestUtils.createTopic(zkClient, Topic.GROUP_METADATA_TOPIC_NAME, OffsetConfig.DefaultOffsetsTopicNumPartitions,
       replicationFactor = 2, servers, servers.head.groupCoordinator.offsetsTopicConfigs)
 
+    createScramCredentials(zkConnect, JaasTestUtils.KafkaScramUser, JaasTestUtils.KafkaScramPassword)
+
     servers.head.config.listeners.foreach { endPoint =>
       val listenerName = endPoint.listenerName
 
-      TestUtils.createTopic(zkClient, listenerName.value, 2, 2, servers)
-
       val trustStoreFile =
         if (TestUtils.usesSslTransportLayer(endPoint.securityProtocol)) Some(this.trustStoreFile)
         else None
 
-      val saslProps =
-        if (TestUtils.usesSaslAuthentication(endPoint.securityProtocol)) Some(saslProperties(listenerName))
-        else None
-
       val bootstrapServers = TestUtils.bootstrapServers(servers, listenerName)
 
-      producers(listenerName) = TestUtils.createNewProducer(bootstrapServers, acks = -1,
-        securityProtocol = endPoint.securityProtocol, trustStoreFile = trustStoreFile, saslProperties = saslProps)
+      def addProducerConsumer(listenerName: ListenerName, mechanism: String, saslProps: Option[Properties]): Unit = {
+
+        val topic = s"${listenerName.value}${producers.size}"
+        TestUtils.createTopic(zkClient, topic, 2, 2, servers)
+        val clientMetadata = ClientMetadata(listenerName, mechanism, topic)
+
+        producers(clientMetadata) = TestUtils.createNewProducer(bootstrapServers, acks = -1,
+          securityProtocol = endPoint.securityProtocol, trustStoreFile = trustStoreFile, saslProperties = saslProps)
+
+        consumers(clientMetadata) = TestUtils.createNewConsumer(bootstrapServers, groupId = clientMetadata.toString,
+          securityProtocol = endPoint.securityProtocol, trustStoreFile = trustStoreFile, saslProperties = saslProps)
+      }
 
-      consumers(listenerName) = TestUtils.createNewConsumer(bootstrapServers, groupId = listenerName.value,
-        securityProtocol = endPoint.securityProtocol, trustStoreFile = trustStoreFile, saslProperties = saslProps)
+      if (TestUtils.usesSaslAuthentication(endPoint.securityProtocol)) {
+        kafkaServerSaslMechanisms(endPoint.listenerName.value).foreach { mechanism =>
+          addProducerConsumer(listenerName, mechanism, Some(kafkaClientSaslProperties(mechanism, dynamicJaasConfig = true)))
+        }
+      } else {
+        addProducerConsumer(listenerName, "", saslProps = None)
+      }
     }
   }
 
@@ -145,18 +162,34 @@ abstract class MultipleListenersWithSameSecurityProtocolBaseTest extends ZooKeep
     */
   @Test
   def testProduceConsume(): Unit = {
-    producers.foreach { case (listenerName, producer) =>
-      val producerRecords = (1 to 10).map(i => new ProducerRecord(listenerName.value, s"key$i".getBytes,
+    producers.foreach { case (clientMetadata, producer) =>
+      val producerRecords = (1 to 10).map(i => new ProducerRecord(clientMetadata.topic, s"key$i".getBytes,
         s"value$i".getBytes))
       producerRecords.map(producer.send).map(_.get(10, TimeUnit.SECONDS))
 
-      val consumer = consumers(listenerName)
-      consumer.subscribe(Collections.singleton(listenerName.value))
+      val consumer = consumers(clientMetadata)
+      consumer.subscribe(Collections.singleton(clientMetadata.topic))
       val records = new ArrayBuffer[ConsumerRecord[Array[Byte], Array[Byte]]]
       TestUtils.waitUntilTrue(() => {
         records ++= consumer.poll(50).asScala
         records.size == producerRecords.size
-      }, s"Consumed ${records.size} records until timeout instead of the expected ${producerRecords.size} records")
+      }, s"Consumed ${records.size} records until timeout instead of the expected ${producerRecords.size} records with mechanism ${clientMetadata.saslMechanism}")
+    }
+  }
+
+  protected def addDynamicJaasSection(props: Properties, listener: String, mechanism: String, jaasSection: JaasSection): Unit = {
+    val listenerName = new ListenerName(listener)
+    val prefix = listenerName.saslMechanismConfigPrefix(mechanism)
+    val jaasConfig = jaasSection.modules.head.toString
+    props.put(s"${prefix}${KafkaConfig.SaslJaasConfigProp}", jaasConfig)
+  }
+
+  case class ClientMetadata(val listenerName: ListenerName, val saslMechanism: String, topic: String) {
+    override def hashCode: Int = Objects.hash(listenerName, saslMechanism)
+    override def equals(obj: Any): Boolean = obj match {
+      case other: ClientMetadata => listenerName == other.listenerName && saslMechanism == other.saslMechanism && topic == other.topic
+      case _ => false
     }
+    override def toString: String = s"${listenerName.value}:$saslMechanism:$topic"
   }
 }
diff --git a/core/src/test/scala/unit/kafka/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
index c934c4a..18be167 100644
--- a/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
@@ -87,13 +87,13 @@ class KafkaTest {
     val config = KafkaConfig.fromProps(Kafka.getPropsFromArgs(Array(propertiesFile, "--override", "ssl.keystore.password=keystore_password",
                                                                                     "--override", "ssl.key.password=key_password",
                                                                                     "--override", "ssl.truststore.password=truststore_password")))
-    assertEquals(Password.HIDDEN, config.sslKeyPassword.toString)
-    assertEquals(Password.HIDDEN, config.sslKeystorePassword.toString)
-    assertEquals(Password.HIDDEN, config.sslTruststorePassword.toString)
+    assertEquals(Password.HIDDEN, config.getPassword(KafkaConfig.SslKeyPasswordProp).toString)
+    assertEquals(Password.HIDDEN, config.getPassword(KafkaConfig.SslKeystorePasswordProp).toString)
+    assertEquals(Password.HIDDEN, config.getPassword(KafkaConfig.SslTruststorePasswordProp).toString)
 
-    assertEquals("key_password", config.sslKeyPassword.value)
-    assertEquals("keystore_password", config.sslKeystorePassword.value)
-    assertEquals("truststore_password", config.sslTruststorePassword.value)
+    assertEquals("key_password", config.getPassword(KafkaConfig.SslKeyPasswordProp).value)
+    assertEquals("keystore_password", config.getPassword(KafkaConfig.SslKeystorePasswordProp).value)
+    assertEquals("truststore_password", config.getPassword(KafkaConfig.SslTruststorePasswordProp).value)
   }
 
   def prepareDefaultConfig(): String = {
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index 6e78423..b3c46fa 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -27,10 +27,10 @@ import kafka.zk.{AdminZkClient, KafkaZkClient, ZooKeeperTestHarness}
 import org.apache.kafka.clients.admin._
 import org.apache.kafka.common.config.ConfigResource
 import org.apache.kafka.common.internals.KafkaFutureImpl
-import org.apache.kafka.common.{KafkaFuture, Node}
+import org.apache.kafka.common.Node
 import org.apache.kafka.common.security.scram.ScramCredentialUtils
 import org.apache.kafka.common.utils.Sanitizer
-import org.easymock.{EasyMock, IAnswer}
+import org.easymock.EasyMock
 import org.junit.Assert._
 import org.junit.Test
 
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index 13299c7..724e05b 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -38,6 +38,7 @@ import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.record.MemoryRecords
 import org.apache.kafka.common.requests.{AbstractRequest, ProduceRequest, RequestHeader}
 import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
+import org.apache.kafka.common.security.scram.ScramMechanism
 import org.apache.kafka.common.utils.{LogContext, MockTime, Time}
 import org.apache.log4j.Level
 import org.junit.Assert._
@@ -61,7 +62,7 @@ class SocketServerTest extends JUnitSuite {
   props.put("connections.max.idle.ms", "60000")
   val config = KafkaConfig.fromProps(props)
   val metrics = new Metrics
-  val credentialProvider = new CredentialProvider(config.saslEnabledMechanisms, null)
+  val credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, null)
   val localAddress = InetAddress.getLoopbackAddress
 
   // Clean-up any metrics left around by previous tests
@@ -406,7 +407,7 @@ class SocketServerTest extends JUnitSuite {
     // the following sleep is necessary to reliably detect the connection close when we send data below
     Thread.sleep(200L)
     // make sure the sockets are open
-    server.acceptors.values.foreach(acceptor => assertFalse(acceptor.serverChannel.socket.isClosed))
+    server.acceptors.asScala.values.foreach(acceptor => assertFalse(acceptor.serverChannel.socket.isClosed))
     // then shutdown the server
     shutdownServerAndMetrics(server)
 
diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
index 6dedbe0..2e3e274 100755
--- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
@@ -17,13 +17,18 @@
 
 package kafka.server
 
+import java.util
 import java.util.Properties
 
 import kafka.utils.TestUtils
+import org.apache.kafka.common.Reconfigurable
+import org.apache.kafka.common.config.types.Password
 import org.apache.kafka.common.config.{ConfigException, SslConfigs}
 import org.junit.Assert._
 import org.junit.Test
 
+import scala.collection.JavaConverters._
+
 class DynamicBrokerConfigTest {
 
   @Test
@@ -34,7 +39,7 @@ class DynamicBrokerConfigTest {
     val config = KafkaConfig(props)
     val dynamicConfig = config.dynamicConfig
     assertSame(config, dynamicConfig.currentKafkaConfig)
-    assertEquals(oldKeystore, config.sslKeystoreLocation)
+    assertEquals(oldKeystore, config.values.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
     assertEquals(oldKeystore,
       config.valuesFromThisConfigWithPrefixOverride("listener.name.external.").get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
     assertEquals(oldKeystore, config.originalsFromThisConfig.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
@@ -55,7 +60,7 @@ class DynamicBrokerConfigTest {
       assertEquals(newKeystore,
         config.originalsWithPrefix("listener.name.external.").get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
 
-      assertEquals(oldKeystore, config.sslKeystoreLocation)
+      assertEquals(oldKeystore, config.getString(KafkaConfig.SslKeystoreLocationProp))
       assertEquals(oldKeystore, config.originals.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
       assertEquals(oldKeystore, config.values.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
       assertEquals(oldKeystore, config.originalsStrings.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
@@ -75,38 +80,48 @@ class DynamicBrokerConfigTest {
     origProps.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "JKS")
     val config = KafkaConfig(origProps)
 
-    def verifyConfigUpdateWithInvalidConfig(validProps: Map[String, String], invalidProps: Map[String, String]): Unit = {
-      val props = new Properties
-      validProps.foreach { case (k, v) => props.put(k, v) }
-      invalidProps.foreach { case (k, v) => props.put(k, v) }
-
-      // DynamicBrokerConfig#validate is used by AdminClient to validate the configs provided in
-      // in an AlterConfigs request. Validation should fail with an exception if any of the configs are invalid.
-      try {
-        config.dynamicConfig.validate(props, perBrokerConfig = true)
-        fail("Invalid config did not fail validation")
-      } catch {
-        case e: ConfigException => // expected exception
-      }
-
-      // DynamicBrokerConfig#updateBrokerConfig is used to update configs from ZooKeeper during
-      // startup and when configs are updated in ZK. Update should apply valid configs and ignore
-      // invalid ones.
-      config.dynamicConfig.updateBrokerConfig(0, props)
-      validProps.foreach { case (name, value) => assertEquals(value, config.originals.get(name)) }
-      invalidProps.keySet.foreach { name =>
-        assertEquals(origProps.get(name), config.originals.get(name))
-      }
-    }
+    val validProps = Map(s"listener.name.external.${SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG}" -> "ks.p12")
 
-    val validProps = Map(s"listener.name.external.${SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG}" ->"ks.p12")
     val securityPropsWithoutListenerPrefix = Map(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG -> "PKCS12")
-    verifyConfigUpdateWithInvalidConfig(validProps, securityPropsWithoutListenerPrefix)
+    verifyConfigUpdateWithInvalidConfig(config, origProps, validProps, securityPropsWithoutListenerPrefix)
     val nonDynamicProps = Map(KafkaConfig.ZkConnectProp -> "somehost:2181")
-    verifyConfigUpdateWithInvalidConfig(validProps, nonDynamicProps)
+    verifyConfigUpdateWithInvalidConfig(config, origProps, validProps, nonDynamicProps)
 
+    // Test update of configs with invalid type
     val invalidProps = Map(KafkaConfig.LogCleanerThreadsProp -> "invalid")
-    verifyConfigUpdateWithInvalidConfig(validProps, invalidProps)
+    verifyConfigUpdateWithInvalidConfig(config, origProps, validProps, invalidProps)
+  }
+
+  @Test
+  def testConfigUpdateWithReconfigurableValidationFailure(): Unit = {
+    val origProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+    origProps.put(KafkaConfig.LogCleanerDedupeBufferSizeProp, "100000000")
+    val config = KafkaConfig(origProps)
+    val validProps = Map.empty[String, String]
+    val invalidProps = Map(KafkaConfig.LogCleanerThreadsProp -> "20")
+
+    def validateLogCleanerConfig(configs: util.Map[String, _]): Unit = {
+      val cleanerThreads = configs.get(KafkaConfig.LogCleanerThreadsProp).toString.toInt
+      if (cleanerThreads <=0 || cleanerThreads >= 5)
+        throw new ConfigException(s"Invalid cleaner threads $cleanerThreads")
+    }
+    val reconfigurable = new Reconfigurable {
+      override def configure(configs: util.Map[String, _]): Unit = {}
+      override def reconfigurableConfigs(): util.Set[String] = Set(KafkaConfig.LogCleanerThreadsProp).asJava
+      override def validateReconfiguration(configs: util.Map[String, _]): Unit = validateLogCleanerConfig(configs)
+      override def reconfigure(configs: util.Map[String, _]): Unit = {}
+    }
+    config.dynamicConfig.addReconfigurable(reconfigurable)
+    verifyConfigUpdateWithInvalidConfig(config, origProps, validProps, invalidProps)
+    config.dynamicConfig.removeReconfigurable(reconfigurable)
+
+    val brokerReconfigurable = new BrokerReconfigurable {
+      override def reconfigurableConfigs: collection.Set[String] = Set(KafkaConfig.LogCleanerThreadsProp)
+      override def validateReconfiguration(newConfig: KafkaConfig): Unit = validateLogCleanerConfig(newConfig.originals)
+      override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {}
+    }
+    config.dynamicConfig.addBrokerReconfigurable(brokerReconfigurable)
+    verifyConfigUpdateWithInvalidConfig(config, origProps, validProps, invalidProps)
   }
 
   @Test
@@ -151,4 +166,86 @@ class DynamicBrokerConfigTest {
       assertEquals(oldValue, config.originals.get(name))
     }
   }
+
+  private def verifyConfigUpdateWithInvalidConfig(config: KafkaConfig,
+                                                  origProps: Properties,
+                                                  validProps: Map[String, String],
+                                                  invalidProps: Map[String, String]): Unit = {
+    val props = new Properties
+    validProps.foreach { case (k, v) => props.put(k, v) }
+    invalidProps.foreach { case (k, v) => props.put(k, v) }
+
+    // DynamicBrokerConfig#validate is used by AdminClient to validate the configs provided in
+    // in an AlterConfigs request. Validation should fail with an exception if any of the configs are invalid.
+    try {
+      config.dynamicConfig.validate(props, perBrokerConfig = true)
+      fail("Invalid config did not fail validation")
+    } catch {
+      case e: ConfigException => // expected exception
+    }
+
+    // DynamicBrokerConfig#updateBrokerConfig is used to update configs from ZooKeeper during
+    // startup and when configs are updated in ZK. Update should apply valid configs and ignore
+    // invalid ones.
+    config.dynamicConfig.updateBrokerConfig(0, props)
+    validProps.foreach { case (name, value) => assertEquals(value, config.originals.get(name)) }
+    invalidProps.keySet.foreach { name =>
+      assertEquals(origProps.get(name), config.originals.get(name))
+    }
+  }
+
+  @Test
+  def testPasswordConfigEncryption(): Unit = {
+    val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+    val configWithoutSecret = KafkaConfig(props)
+    props.put(KafkaConfig.PasswordEncoderSecretProp, "config-encoder-secret")
+    val configWithSecret = KafkaConfig(props)
+    val dynamicProps = new Properties
+    dynamicProps.put(KafkaConfig.SaslJaasConfigProp, "myLoginModule required;")
+
+    try {
+      configWithoutSecret.dynamicConfig.toPersistentProps(dynamicProps, perBrokerConfig = true)
+    } catch {
+      case e: ConfigException => // expected exception
+    }
+    val persistedProps = configWithSecret.dynamicConfig.toPersistentProps(dynamicProps, perBrokerConfig = true)
+    assertFalse("Password not encoded",
+      persistedProps.getProperty(KafkaConfig.SaslJaasConfigProp).contains("myLoginModule"))
+    val decodedProps = configWithSecret.dynamicConfig.fromPersistentProps(persistedProps, perBrokerConfig = true)
+    assertEquals("myLoginModule required;", decodedProps.getProperty(KafkaConfig.SaslJaasConfigProp))
+  }
+
+  @Test
+  def testPasswordConfigEncoderSecretChange(): Unit = {
+    val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+    props.put(KafkaConfig.SaslJaasConfigProp, "staticLoginModule required;")
+    props.put(KafkaConfig.PasswordEncoderSecretProp, "config-encoder-secret")
+    val config = KafkaConfig(props)
+    val dynamicProps = new Properties
+    dynamicProps.put(KafkaConfig.SaslJaasConfigProp, "dynamicLoginModule required;")
+
+    val persistedProps = config.dynamicConfig.toPersistentProps(dynamicProps, perBrokerConfig = true)
+    assertFalse("Password not encoded",
+      persistedProps.getProperty(KafkaConfig.SaslJaasConfigProp).contains("LoginModule"))
+    config.dynamicConfig.updateBrokerConfig(0, persistedProps)
+    assertEquals("dynamicLoginModule required;", config.values.get(KafkaConfig.SaslJaasConfigProp).asInstanceOf[Password].value)
+
+    // New config with same secret should use the dynamic password config
+    val newConfigWithSameSecret = KafkaConfig(props)
+    newConfigWithSameSecret.dynamicConfig.updateBrokerConfig(0, persistedProps)
+    assertEquals("dynamicLoginModule required;", newConfigWithSameSecret.values.get(KafkaConfig.SaslJaasConfigProp).asInstanceOf[Password].value)
+
+    // New config with new secret should use the dynamic password config if new and old secrets are configured in KafkaConfig
+    props.put(KafkaConfig.PasswordEncoderSecretProp, "new-encoder-secret")
+    props.put(KafkaConfig.PasswordEncoderOldSecretProp, "config-encoder-secret")
+    val newConfigWithNewAndOldSecret = KafkaConfig(props)
+    newConfigWithNewAndOldSecret.dynamicConfig.updateBrokerConfig(0, persistedProps)
+    assertEquals("dynamicLoginModule required;", newConfigWithSameSecret.values.get(KafkaConfig.SaslJaasConfigProp).asInstanceOf[Password].value)
+
+    // New config with new secret alone should revert to static password config since dynamic config cannot be decoded
+    props.put(KafkaConfig.PasswordEncoderSecretProp, "another-new-encoder-secret")
+    val newConfigWithNewSecret = KafkaConfig(props)
+    newConfigWithNewSecret.dynamicConfig.updateBrokerConfig(0, persistedProps)
+    assertEquals("staticLoginModule required;", newConfigWithNewSecret.values.get(KafkaConfig.SaslJaasConfigProp).asInstanceOf[Password].value)
+  }
 }
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 748efec..6b26334 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -674,12 +674,22 @@ class KafkaConfigTest {
         case KafkaConfig.SaslKerberosTicketRenewJitterProp =>
         case KafkaConfig.SaslKerberosMinTimeBeforeReloginProp =>
         case KafkaConfig.SaslKerberosPrincipalToLocalRulesProp => // ignore string
+        case KafkaConfig.SaslJaasConfigProp =>
+
+        // Password encoder configs
+        case KafkaConfig.PasswordEncoderSecretProp =>
+        case KafkaConfig.PasswordEncoderOldSecretProp =>
+        case KafkaConfig.PasswordEncoderKeyFactoryAlgorithmProp =>
+        case KafkaConfig.PasswordEncoderCipherAlgorithmProp =>
+        case KafkaConfig.PasswordEncoderKeyLengthProp => assertPropertyInvalid(getBaseProperties, name, "not_a_number", "-1", "0")
+        case KafkaConfig.PasswordEncoderIterationsProp => assertPropertyInvalid(getBaseProperties, name, "not_a_number", "-1", "0")
 
         //delegation token configs
         case KafkaConfig.DelegationTokenMasterKeyProp => // ignore
         case KafkaConfig.DelegationTokenMaxLifeTimeProp => assertPropertyInvalid(getBaseProperties(), name, "not_a_number", "0")
         case KafkaConfig.DelegationTokenExpiryTimeMsProp => assertPropertyInvalid(getBaseProperties(), name, "not_a_number", "0")
         case KafkaConfig.DelegationTokenExpiryCheckIntervalMsProp => assertPropertyInvalid(getBaseProperties(), name, "not_a_number", "0")
+
         case _ => assertPropertyInvalid(getBaseProperties(), name, "not_a_number", "-1")
       }
     })
diff --git a/core/src/test/scala/unit/kafka/utils/PasswordEncoderTest.scala b/core/src/test/scala/unit/kafka/utils/PasswordEncoderTest.scala
new file mode 100755
index 0000000..11a2a7a
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/utils/PasswordEncoderTest.scala
@@ -0,0 +1,127 @@
+/**
+ * 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 kafka.utils
+
+
+import javax.crypto.SecretKeyFactory
+
+import kafka.server.Defaults
+import org.apache.kafka.common.config.ConfigException
+import org.apache.kafka.common.config.types.Password
+import org.apache.kafka.common.utils.Java
+import org.junit.Assert._
+import org.junit.Test
+
+class PasswordEncoderTest {
+
+  @Test
+  def testEncodeDecode(): Unit = {
+    val encoder = new PasswordEncoder(new Password("password-encoder-secret"),
+      None,
+      Defaults.PasswordEncoderCipherAlgorithm,
+      Defaults.PasswordEncoderKeyLength,
+      Defaults.PasswordEncoderIterations)
+    val password = "test-password"
+    val encoded = encoder.encode(new Password(password))
+    val encodedMap = CoreUtils.parseCsvMap(encoded)
+    assertEquals("4096", encodedMap(PasswordEncoder.IterationsProp))
+    assertEquals("128", encodedMap(PasswordEncoder.KeyLengthProp))
+    val defaultKeyFactoryAlgorithm = try {
+      SecretKeyFactory.getInstance("PBKDF2WithHmacSHA512")
+      "PBKDF2WithHmacSHA512"
+    } catch {
+      case _: Exception => "PBKDF2WithHmacSHA1"
+    }
+    assertEquals(defaultKeyFactoryAlgorithm, encodedMap(PasswordEncoder.KeyFactoryAlgorithmProp))
+    assertEquals("AES/CBC/PKCS5Padding", encodedMap(PasswordEncoder.CipherAlgorithmProp))
+
+    verifyEncodedPassword(encoder, password, encoded)
+  }
+
+  @Test
+  def testEncoderConfigChange(): Unit = {
+    val encoder = new PasswordEncoder(new Password("password-encoder-secret"),
+      Some("PBKDF2WithHmacSHA1"),
+      "DES/CBC/PKCS5Padding",
+      64,
+      1024)
+    val password = "test-password"
+    val encoded = encoder.encode(new Password(password))
+    val encodedMap = CoreUtils.parseCsvMap(encoded)
+    assertEquals("1024", encodedMap(PasswordEncoder.IterationsProp))
+    assertEquals("64", encodedMap(PasswordEncoder.KeyLengthProp))
+    assertEquals("PBKDF2WithHmacSHA1", encodedMap(PasswordEncoder.KeyFactoryAlgorithmProp))
+    assertEquals("DES/CBC/PKCS5Padding", encodedMap(PasswordEncoder.CipherAlgorithmProp))
+
+    // Test that decoding works even if PasswordEncoder algorithm, iterations etc. are altered
+    val decoder = new PasswordEncoder(new Password("password-encoder-secret"),
+      Some("PBKDF2WithHmacSHA1"),
+      "AES/CBC/PKCS5Padding",
+      128,
+      2048)
+    assertEquals(password, decoder.decode(encoded).value)
+
+    // Test that decoding fails if secret is altered
+    val decoder2 = new PasswordEncoder(new Password("secret-2"),
+      Some("PBKDF2WithHmacSHA1"),
+      "AES/CBC/PKCS5Padding",
+      128,
+      1024)
+    try {
+      decoder2.decode(encoded)
+    } catch {
+      case e: ConfigException => // expected exception
+    }
+  }
+
+  @Test
+  def testEncodeDecodeAlgorithms(): Unit = {
+
+    def verifyEncodeDecode(keyFactoryAlg: Option[String], cipherAlg: String, keyLength: Int): Unit = {
+      val encoder = new PasswordEncoder(new Password("password-encoder-secret"),
+        keyFactoryAlg,
+        cipherAlg,
+        keyLength,
+        Defaults.PasswordEncoderIterations)
+      val password = "test-password"
+      val encoded = encoder.encode(new Password(password))
+      verifyEncodedPassword(encoder, password, encoded)
+    }
+
+    verifyEncodeDecode(keyFactoryAlg = None, "DES/CBC/PKCS5Padding", keyLength = 64)
+    verifyEncodeDecode(keyFactoryAlg = None, "DESede/CBC/PKCS5Padding", keyLength = 192)
+    verifyEncodeDecode(keyFactoryAlg = None, "AES/CBC/PKCS5Padding", keyLength = 128)
+    verifyEncodeDecode(keyFactoryAlg = None, "AES/CFB/PKCS5Padding", keyLength = 128)
+    verifyEncodeDecode(keyFactoryAlg = None, "AES/OFB/PKCS5Padding", keyLength = 128)
+    verifyEncodeDecode(keyFactoryAlg = Some("PBKDF2WithHmacSHA1"), Defaults.PasswordEncoderCipherAlgorithm, keyLength = 128)
+    if (Java.IS_JAVA8_COMPATIBLE) {
+      verifyEncodeDecode(keyFactoryAlg = None, "AES/GCM/PKCS5Padding", keyLength = 128)
+      verifyEncodeDecode(keyFactoryAlg = Some("PBKDF2WithHmacSHA256"), Defaults.PasswordEncoderCipherAlgorithm, keyLength = 128)
+      verifyEncodeDecode(keyFactoryAlg = Some("PBKDF2WithHmacSHA512"), Defaults.PasswordEncoderCipherAlgorithm, keyLength = 128)
+    }
+  }
+
+  private def verifyEncodedPassword(encoder: PasswordEncoder, password: String, encoded: String): Unit = {
+    val encodedMap = CoreUtils.parseCsvMap(encoded)
+    assertEquals(password.length.toString, encodedMap(PasswordEncoder.PasswordLengthProp))
+    assertNotNull("Invalid salt", encoder.base64Decode(encodedMap("salt")))
+    assertNotNull("Invalid encoding parameters", encoder.base64Decode(encodedMap(PasswordEncoder.InitializationVectorProp)))
+    assertNotNull("Invalid encoded password", encoder.base64Decode(encodedMap(PasswordEncoder.EncyrptedPasswordProp)))
+    assertEquals(password, encoder.decode(encoded).value)
+  }
+}
\ No newline at end of file
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 407bdb5..303afa7 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -718,7 +718,7 @@ object TestUtils extends Logging {
 
   def deleteBrokersInZk(zkClient: KafkaZkClient, ids: Seq[Int]): Seq[Broker] = {
     val brokers = ids.map(createBroker(_, "localhost", 6667, SecurityProtocol.PLAINTEXT))
-    brokers.foreach(b => zkClient.deletePath(BrokerIdsZNode.path + "/" + b))
+    ids.foreach(b => zkClient.deletePath(BrokerIdsZNode.path + "/" + b))
     brokers
   }
 

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

Mime
View raw message