MINOR: Improved code quality for various files. (#9037)

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>

clients/src/main/java/org/apache/kafka/common/security/JaasContext.java

@@ -62,12 +62,11 @@ public class JaasContext {
             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 dynamicJaasConfig = (Password) configs.get(mechanism.toLowerCase(Locale.ROOT) + "." + SaslConfigs.SASL_JAAS_CONFIG);
         if (dynamicJaasConfig == null && configs.get(SaslConfigs.SASL_JAAS_CONFIG) != null)
             LOG.warn("Server config {} should be prefixed with SASL mechanism name, ignoring config", SaslConfigs.SASL_JAAS_CONFIG);
-        return load(Type.SERVER, listenerContextName, globalContextName, dynamicJaasConfig);
+        return load(Type.SERVER, listenerContextName, GLOBAL_CONTEXT_NAME_SERVER, dynamicJaasConfig);
     }
 
     /**
@@ -79,9 +78,8 @@ public class JaasContext {
      *
      */
     public static JaasContext loadClientContext(Map<String, ?> configs) {
-        String globalContextName = GLOBAL_CONTEXT_NAME_CLIENT;
         Password dynamicJaasConfig = (Password) configs.get(SaslConfigs.SASL_JAAS_CONFIG);
-        return load(JaasContext.Type.CLIENT, null, globalContextName, dynamicJaasConfig);
+        return load(JaasContext.Type.CLIENT, null, GLOBAL_CONTEXT_NAME_CLIENT, dynamicJaasConfig);
     }
 
     static JaasContext load(JaasContext.Type contextType, String listenerContextName,

clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java

@@ -672,7 +672,7 @@ public class SaslClientAuthenticator implements Authenticator {
                 double pctWindowJitterToAvoidReauthenticationStormAcrossManyChannelsSimultaneously = 0.10;
                 double pctToUse = pctWindowFactorToTakeNetworkLatencyAndClockDriftIntoAccount + RNG.nextDouble()
                         * pctWindowJitterToAvoidReauthenticationStormAcrossManyChannelsSimultaneously;
-                sessionLifetimeMsToUse = (long) (positiveSessionLifetimeMs.longValue() * pctToUse);
+                sessionLifetimeMsToUse = (long) (positiveSessionLifetimeMs * pctToUse);
                 clientSessionReauthenticationTimeNanos = authenticationEndNanos + 1000 * 1000 * sessionLifetimeMsToUse;
                 log.debug(
                         "Finished {} with session expiration in {} ms and session re-authentication on or after {} ms",
@@ -684,7 +684,7 @@ public class SaslClientAuthenticator implements Authenticator {
 
         public Long reauthenticationLatencyMs() {
             return reauthenticating()
-                    ? Long.valueOf(Math.round((authenticationEndNanos - reauthenticationBeginNanos) / 1000.0 / 1000.0))
+                    ? Math.round((authenticationEndNanos - reauthenticationBeginNanos) / 1000.0 / 1000.0)
                     : null;
         }
 

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

@@ -115,14 +115,14 @@ public final class DefaultSslEngineFactory implements SslEngineFactory {
 
         List<String> cipherSuitesList = (List<String>) configs.get(SslConfigs.SSL_CIPHER_SUITES_CONFIG);
         if (cipherSuitesList != null && !cipherSuitesList.isEmpty()) {
-            this.cipherSuites = cipherSuitesList.toArray(new String[cipherSuitesList.size()]);
+            this.cipherSuites = cipherSuitesList.toArray(new String[0]);
         } else {
             this.cipherSuites = null;
         }
 
         List<String> enabledProtocolsList = (List<String>) configs.get(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG);
         if (enabledProtocolsList != null && !enabledProtocolsList.isEmpty()) {
-            this.enabledProtocols = enabledProtocolsList.toArray(new String[enabledProtocolsList.size()]);
+            this.enabledProtocols = enabledProtocolsList.toArray(new String[0]);
         } else {
             this.enabledProtocols = null;
         }
@@ -149,7 +149,7 @@ public final class DefaultSslEngineFactory implements SslEngineFactory {
     }
 
     @Override
-    public void close() throws IOException {
+    public void close() {
         this.sslContext = null;
     }
 
@@ -194,7 +194,7 @@ public final class DefaultSslEngineFactory implements SslEngineFactory {
         log.warn("Unrecognized client authentication configuration {}.  Falling " +
                 "back to NONE.  Recognized client authentication configurations are {}.",
                 key, String.join(", ", SslClientAuth.VALUES.stream().
-                        map(a -> a.name()).collect(Collectors.toList())));
+                        map(Enum::name).collect(Collectors.toList())));
         return SslClientAuth.NONE;
     }
 

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

@@ -163,8 +163,7 @@ public class SslPrincipalMapper {
                 if (backRefNum.startsWith("0")) {
                     continue;
                 }
-                final int originalBackRefIndex = Integer.parseInt(backRefNum);
-                int backRefIndex = originalBackRefIndex;
+                int backRefIndex = Integer.parseInt(backRefNum);
 
 
                 // if we have a replacement value like $123, and we have less than 123 capturing groups, then