More refactoring

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientCredentialsAccessTokenRetriever.java

@@ -67,7 +67,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
         super.configure(configs, saslMechanism, jaasConfigEntries);
 
         JaasOptionsUtils jou = new JaasOptionsUtils(saslMechanism, jaasConfigEntries);
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         clientId = jou.validateString(CLIENT_ID_CONFIG);
         clientSecret = jou.validateString(CLIENT_SECRET_CONFIG);
         scope = jou.validateString(SCOPE_CONFIG, false);

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtils.java

@@ -45,11 +45,7 @@ public class ConfigurationUtils {
 
     private final String prefix;
 
-    public ConfigurationUtils(Map<String, ?> configs) {
-        this(null, configs);
-    }
-
-    public ConfigurationUtils(String saslMechanism, Map<String, ?> configs) {
+    public ConfigurationUtils(Map<String, ?> configs, String saslMechanism) {
         this.configs = configs;
 
         if (saslMechanism != null && !saslMechanism.trim().isEmpty())

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultAccessTokenRetriever.java

@@ -32,11 +32,11 @@
 
 public class DefaultAccessTokenRetriever implements AccessTokenRetriever {
 
-    protected AccessTokenRetriever delegate;
+    private AccessTokenRetriever delegate;
 
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         URL tokenEndpointUrl = cu.validateUrl(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL);
 
         if (tokenEndpointUrl.getProtocol().toLowerCase(Locale.ROOT).equals("file")) {

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultAccessTokenValidator.java

@@ -55,7 +55,7 @@
 
 public class DefaultAccessTokenValidator implements AccessTokenValidator {
 
-    protected AccessTokenValidator delegate;
+    private AccessTokenValidator delegate;
 
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DelegatingVerificationKeyResolver.java

@@ -47,7 +47,7 @@ public DelegatingVerificationKeyResolver(Time time) {
 
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         URL jwksEndpointUrl = cu.validateUrl(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL);
 
         if (jwksEndpointUrl.getProtocol().toLowerCase(Locale.ROOT).equals("file"))

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/FileTokenRetriever.java

@@ -38,11 +38,11 @@
 
 public class FileTokenRetriever implements AccessTokenRetriever {
 
-    protected String accessToken;
+    private String accessToken;
 
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         String accessTokenFileName = cu.validateFile(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL).toFile().getPath();
 
         try {

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/HttpAccessTokenRetriever.java

@@ -18,7 +18,6 @@
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.config.SaslConfigs;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -41,12 +40,10 @@
 import javax.net.ssl.SSLSocketFactory;
 import javax.security.auth.login.AppConfigurationEntry;
 
-import static org.apache.kafka.common.config.SaslConfigs.DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_LOGIN_CONNECT_TIMEOUT_MS;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_LOGIN_READ_TIMEOUT_MS;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_LOGIN_RETRY_BACKOFF_MAX_MS;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_LOGIN_RETRY_BACKOFF_MS;
-import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL;
 
 /**
@@ -100,7 +97,7 @@ public abstract class HttpAccessTokenRetriever implements AccessTokenRetriever {
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
         JaasOptionsUtils jou = new JaasOptionsUtils(saslMechanism, jaasConfigEntries);
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
 
         URL url = cu.validateUrl(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL);
         tokenEndpointUrl = url.toString();
@@ -279,24 +276,4 @@ static String parseAccessToken(String responseBody) throws IOException {
 
         return value;
     }
-
-    /**
-     * In some cases, the incoming {@link Map} doesn't contain a value for
-     * {@link SaslConfigs#SASL_OAUTHBEARER_HEADER_URLENCODE}. Returning {@code null} from {@link Map#get(Object)}
-     * will cause a {@link NullPointerException} when it is later unboxed.
-     *
-     * <p/>
-     *
-     * This utility method ensures that we have a non-{@code null} value to use in the
-     * {@link HttpAccessTokenRetriever} constructor.
-     */
-    public static boolean validateUrlencodeHeader(ConfigurationUtils configurationUtils) {
-        Boolean urlencodeHeader = configurationUtils.validateBoolean(SASL_OAUTHBEARER_HEADER_URLENCODE, false);
-
-        if (urlencodeHeader != null)
-            return urlencodeHeader;
-        else
-            return DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE;
-    }
-
 }

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JwksFileVerificationKeyResolver.java

@@ -91,7 +91,7 @@ public class JwksFileVerificationKeyResolver implements CloseableVerificationKey
 
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         String jwksFileName = cu.validateFile(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL).toFile().getPath();
         log.debug("Starting creation of new VerificationKeyResolver from {}", jwksFileName);
         JsonWebKeySet jwks;

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JwtBearerAccessTokenRetriever.java

@@ -43,17 +43,20 @@ public class JwtBearerAccessTokenRetriever extends HttpAccessTokenRetriever {
 
     public final static String GRANT_TYPE = "urn:ietf:params:oauth:grant-type:jwt-bearer";
 
+    private final Time time;
+
     private String assertion;
 
-    @Override
-    public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
-        configure(Time.SYSTEM, configs, saslMechanism, jaasConfigEntries);
+    public JwtBearerAccessTokenRetriever() {
+        this(Time.SYSTEM);
     }
 
-    public void configure(Time time,
-                          Map<String, ?> configs,
-                          String saslMechanism,
-                          List<AppConfigurationEntry> jaasConfigEntries) {
+    public JwtBearerAccessTokenRetriever(Time time) {
+        this.time = time;
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
         super.configure(configs, saslMechanism, jaasConfigEntries);
 
         JaasOptionsUtils jou = new JaasOptionsUtils(saslMechanism, jaasConfigEntries);

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/LoginAccessTokenValidator.java

@@ -63,13 +63,13 @@ public class LoginAccessTokenValidator implements AccessTokenValidator {
 
     public static final String ISSUED_AT_CLAIM_NAME = "iat";
 
-    protected String scopeClaimName;
+    private String scopeClaimName;
 
-    protected String subClaimName;
+    private String subClaimName;
 
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         scopeClaimName = ClaimValidationUtils.validateClaimNameOverride(
             DEFAULT_SASL_OAUTHBEARER_SCOPE_CLAIM_NAME,
             cu.get(SASL_OAUTHBEARER_SCOPE_CLAIM_NAME)

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/RefreshingHttpsJwks.java

@@ -150,7 +150,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
         try {
             log.debug("init started");
 
-            ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+            ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
             URL jwksEndpointUrl = cu.validateUrl(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL);
 
             JaasOptionsUtils jou = new JaasOptionsUtils(saslMechanism, jaasConfigEntries);
@@ -177,7 +177,7 @@ void configure(HttpsJwks httpsJwks,
                    Map<String, ?> configs,
                    String saslMechanism,
                    List<AppConfigurationEntry> jaasConfigEntries) {
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         refreshMs = cu.validateLong(SASL_OAUTHBEARER_JWKS_ENDPOINT_REFRESH_MS, true, 0L);
         refreshRetryBackoffMs = cu.validateLong(SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS);
         refreshRetryBackoffMaxMs = cu.validateLong(SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS);

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/RefreshingHttpsJwksVerificationKeyResolver.java

@@ -91,13 +91,13 @@ public class RefreshingHttpsJwksVerificationKeyResolver implements CloseableVeri
 
     private static final Logger log = LoggerFactory.getLogger(RefreshingHttpsJwksVerificationKeyResolver.class);
 
-    protected final Time time;
+    private final Time time;
 
-    protected final VerificationJwkSelector verificationJwkSelector;
+    private final VerificationJwkSelector verificationJwkSelector;
 
-    protected RefreshingHttpsJwks refreshingHttpsJwks;
+    private RefreshingHttpsJwks refreshingHttpsJwks;
 
-    protected boolean isInitialized;
+    private boolean isInitialized;
 
     public RefreshingHttpsJwksVerificationKeyResolver(Time time) {
         this.time = time;

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ValidatorAccessTokenValidator.java

@@ -95,13 +95,13 @@ public class ValidatorAccessTokenValidator implements AccessTokenValidator {
 
     private final Time time;
 
-    protected CloseableVerificationKeyResolver verificationKeyResolver;
+    private CloseableVerificationKeyResolver verificationKeyResolver;
 
-    protected JwtConsumer jwtConsumer;
+    private JwtConsumer jwtConsumer;
 
-    protected String scopeClaimName;
+    private String scopeClaimName;
 
-    protected String subClaimName;
+    private String subClaimName;
 
     public ValidatorAccessTokenValidator() {
         this(Time.SYSTEM);
@@ -134,7 +134,7 @@ void configure(CloseableVerificationKeyResolver verificationKeyResolver,
         this.verificationKeyResolver = verificationKeyResolver;
         this.verificationKeyResolver.configure(configs, saslMechanism, jaasConfigEntries);
 
-        ConfigurationUtils cu = new ConfigurationUtils(saslMechanism, configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
         Set<String> expectedAudiences = null;
         List<String> l = cu.get(SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
 

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientCredentialsAccessTokenRetrieverTest.java

@@ -18,7 +18,17 @@
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
+import java.util.Collections;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.apache.kafka.common.config.SaslConfigs.DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.OAUTHBEARER_MECHANISM;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -84,6 +94,23 @@ public void testFormatRequestBodyMissingValues() {
         assertEquals(expected, actual);
     }
 
+    @ParameterizedTest
+    @MethodSource("urlencodeHeaderSupplier")
+    public void testUrlencodeHeader(Map<String, Object> configs, boolean expectedValue) {
+        ConfigurationUtils cu = new ConfigurationUtils(configs, OAUTHBEARER_MECHANISM);
+        boolean actualValue = ClientCredentialsAccessTokenRetriever.validateUrlencodeHeader(cu);
+        assertEquals(expectedValue, actualValue);
+    }
+
+    private static Stream<Arguments> urlencodeHeaderSupplier() {
+        return Stream.of(
+            Arguments.of(Collections.emptyMap(), DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE),
+            Arguments.of(Collections.singletonMap(SASL_OAUTHBEARER_HEADER_URLENCODE, null), DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE),
+            Arguments.of(Collections.singletonMap(SASL_OAUTHBEARER_HEADER_URLENCODE, true), true),
+            Arguments.of(Collections.singletonMap(SASL_OAUTHBEARER_HEADER_URLENCODE, false), false)
+        );
+    }
+
     private void assertAuthorizationHeader(String clientId, String clientSecret, boolean urlencode, String expected) {
         String actual = ClientCredentialsAccessTokenRetriever.formatAuthorizationHeader(clientId, clientSecret, urlencode);
         assertEquals(expected, actual, String.format("Expected the HTTP Authorization header generated for client ID \"%s\" and client secret \"%s\" to match", clientId, clientSecret));

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtilsTest.java

@@ -30,6 +30,7 @@
 import java.util.Map;
 
 import static org.apache.kafka.common.config.internals.BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.OAUTHBEARER_MECHANISM;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 
 public class ConfigurationUtilsTest extends OAuthBearerTest {
@@ -95,7 +96,7 @@ public void testUrlWhitespace() {
     private void testUrl(String value) {
         System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, value == null ? "" : value);
         Map<String, Object> configs = Collections.singletonMap(URL_CONFIG_NAME, value);
-        ConfigurationUtils cu = new ConfigurationUtils(configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, OAUTHBEARER_MECHANISM);
         cu.validateUrl(URL_CONFIG_NAME);
     }
 
@@ -148,7 +149,7 @@ public void testThrowIfURLIsNotAllowed() {
         Map<String, Object> configs = new HashMap<>();
         configs.put(URL_CONFIG_NAME, url);
         configs.put(FILE_CONFIG_NAME, fileUrl);
-        ConfigurationUtils cu = new ConfigurationUtils(configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, OAUTHBEARER_MECHANISM);
 
         // By default, no URL is allowed
         assertThrowsWithMessage(ConfigException.class, () -> cu.throwIfURLIsNotAllowed(url),
@@ -171,7 +172,7 @@ public void testThrowIfURLIsNotAllowed() {
     protected void testFile(String value) {
         System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, value == null ? "" : value);
         Map<String, Object> configs = Collections.singletonMap(URL_CONFIG_NAME, value);
-        ConfigurationUtils cu = new ConfigurationUtils(configs);
+        ConfigurationUtils cu = new ConfigurationUtils(configs, OAUTHBEARER_MECHANISM);
         cu.validateFile(URL_CONFIG_NAME);
     }
 

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/HttpAccessTokenRetrieverTest.java

@@ -21,20 +21,13 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
 
 import java.io.IOException;
 import java.net.HttpURLConnection;
 import java.nio.charset.StandardCharsets;
-import java.util.Collections;
-import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Stream;
 
-import static org.apache.kafka.common.config.SaslConfigs.DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE;
-import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -121,23 +114,6 @@ public void testParseAccessTokenInvalidJson() {
         assertThrows(IOException.class, () -> HttpAccessTokenRetriever.parseAccessToken("not valid JSON"));
     }
 
-    @ParameterizedTest
-    @MethodSource("urlencodeHeaderSupplier")
-    public void testUrlencodeHeader(Map<String, Object> configs, boolean expectedValue) {
-        ConfigurationUtils cu = new ConfigurationUtils(configs);
-        boolean actualValue = HttpAccessTokenRetriever.validateUrlencodeHeader(cu);
-        assertEquals(expectedValue, actualValue);
-    }
-
-    private static Stream<Arguments> urlencodeHeaderSupplier() {
-        return Stream.of(
-            Arguments.of(Collections.emptyMap(), DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE),
-            Arguments.of(Collections.singletonMap(SASL_OAUTHBEARER_HEADER_URLENCODE, null), DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE),
-            Arguments.of(Collections.singletonMap(SASL_OAUTHBEARER_HEADER_URLENCODE, true), true),
-            Arguments.of(Collections.singletonMap(SASL_OAUTHBEARER_HEADER_URLENCODE, false), false)
-        );
-    }
-
     private <T extends Exception> void testErrorResponse(Class<T> exceptionClazz,
                                                          int responseCode,
                                                          String errorResponse,