MINOR: Fix encoder config to make DynamicBrokerReconfigurationTest stable (#4764)

DynamicBrokerReconfigurationTest currently assumes that passwords encoded with one secret will fail with an exception if decoded with another secret and configures an old.secret in setUp. This could potentially cause test failures if a password was incorrectly decoded with the wrong secret, since the test writes passwords encoded with the new secret directly to ZooKeeper. Since old.secret is only used in one test for verifying secret rotation, this config can be moved to that test to avoid transient failures.

Reviewers: Ismael Juma <ismael@juma.me.uk>

clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java

@@ -287,17 +287,21 @@ public class SslFactory implements Reconfigurable {
             this.keyPassword = keyPassword;
         }
 
-        KeyStore load() throws GeneralSecurityException, IOException {
-            FileInputStream in = null;
-            try {
+        /**
+         * Loads this keystore
+         * @return the keystore
+         * @throws KafkaException if the file could not be read or if the keystore could not be loaded
+         *   using the specified configs (e.g. if the password or keystore type is invalid)
+         */
+        KeyStore load() {
+            try (FileInputStream in = new FileInputStream(path)) {
                 KeyStore ks = KeyStore.getInstance(type);
-                in = new FileInputStream(path);
                 // If a password is not set access to the truststore is still available, but integrity checking is disabled.
                 char[] passwordChars = password != null ? password.value().toCharArray() : null;
                 ks.load(in, passwordChars);
                 return ks;
-            } finally {
-                if (in != null) in.close();
+            } catch (GeneralSecurityException | IOException e) {
+                throw new KafkaException("Failed to load SSL keystore " + path + " of type " + type, e);
             }
         }
     }

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

@@ -111,7 +111,6 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
       props.put(KafkaConfig.NumReplicaFetchersProp, "2") // greater than one to test reducing threads
       props.put(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp, "10000000") // non-default value to trigger a new metric
       props.put(KafkaConfig.PasswordEncoderSecretProp, "dynamic-config-secret")
-      props.put(KafkaConfig.PasswordEncoderOldSecretProp, "old-dynamic-config-secret") // for testing secret rotation
 
       props ++= sslProperties1
       addKeystoreWithListenerPrefix(sslProperties1, props, SecureInternal)
@@ -670,7 +669,8 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
       val propsEncodedWithOldSecret = props.clone().asInstanceOf[Properties]
       val config = server.config
       val secret = config.passwordEncoderSecret.getOrElse(throw new IllegalStateException("Password encoder secret not configured"))
-      val oldSecret = config.passwordEncoderOldSecret.getOrElse(throw new IllegalStateException("Password encoder old secret not configured"))
+      val oldSecret = "old-dynamic-config-secret"
+      config.dynamicConfig.staticBrokerConfigs.put(KafkaConfig.PasswordEncoderOldSecretProp, oldSecret)
       val passwordConfigs = props.asScala.filterKeys(DynamicBrokerConfig.isPasswordConfig)
       assertTrue("Password configs not found", passwordConfigs.nonEmpty)
       val passwordDecoder = new PasswordEncoder(secret,
@@ -678,7 +678,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
         config.passwordEncoderCipherAlgorithm,
         config.passwordEncoderKeyLength,
         config.passwordEncoderIterations)
-      val passwordEncoder = new PasswordEncoder(oldSecret,
+      val passwordEncoder = new PasswordEncoder(new Password(oldSecret),
         config.passwordEncoderKeyFactoryAlgorithm,
         config.passwordEncoderCipherAlgorithm,
         config.passwordEncoderKeyLength,