This is an automated email from the ASF dual-hosted git repository.
omkreddy pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new ab4bb0a64b3 MINOR: Clean up SASL/OAUTHBEARER expected issuer/audience
config docs and metadata
ab4bb0a64b3 is described below
commit ab4bb0a64b3c0a768c8282042d2bb4ef090a65cc
Author: Evan Zhou <[email protected]>
AuthorDate: Wed Jun 10 16:42:09 2026 -0700
MINOR: Clean up SASL/OAUTHBEARER expected issuer/audience config docs and
metadata
Minor cleanup of the broker-side SASL/OAUTHBEARER validation configuration:
- Clarify the documentation for sasl.oauthbearer.expected.issuer and
sasl.oauthbearer.expected.audience.
- Set their Importance to HIGH, consistently across SaslConfigs and
BrokerSecurityConfigs.
- Log a startup warning when sasl.oauthbearer.expected.issuer is not set,
as a
configuration hint for operators.
- Tidy up the BrokerJwtValidator Javadoc and add related unit tests.
---
checkstyle/import-control.xml | 2 +
.../apache/kafka/common/config/SaslConfigs.java | 12 ++--
.../config/internals/BrokerSecurityConfigs.java | 5 +-
.../security/oauthbearer/BrokerJwtValidator.java | 16 ++++-
.../oauthbearer/BrokerJwtValidatorTest.java | 73 +++++++++++++++++++++-
.../oauthbearer/DefaultJwtValidatorTest.java | 11 +++-
.../kafka/api/ClientOAuthIntegrationTest.scala | 1 +
7 files changed, 105 insertions(+), 15 deletions(-)
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index f8ff30c3eb9..ef840bd8a75 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -140,6 +140,8 @@
<allow pkg="org.testcontainers" />
<allow pkg="org.keycloak" />
<allow pkg="dasniko.testcontainers" />
+ <!-- For asserting log output in tests via LogCaptureAppender -->
+ <allow class="org.apache.logging.log4j.Level" />
</subpackage>
</subpackage>
diff --git
a/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
b/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
index 01f7ad1f927..5f426671110 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
@@ -353,14 +353,14 @@ public class SaslConfigs {
+ " the broker.";
public static final String SASL_OAUTHBEARER_EXPECTED_AUDIENCE =
"sasl.oauthbearer.expected.audience";
- public static final String SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC = "The
(optional) comma-delimited setting for the broker to use to verify that the JWT
was issued for one of the"
+ public static final String SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC = "The
(strongly recommended) comma-delimited setting for the broker to use to verify
that the JWT was issued for one of the"
+ " expected audiences. The JWT will be inspected for the standard
OAuth \"aud\" claim and if this value is set, the broker will match the value
from JWT's \"aud\" claim "
- + " to see if there is an exact match. If there is no match, the
broker will reject the JWT and authentication will fail.";
+ + " to see if there is an exact match. If there is no match, the
broker will reject the JWT and authentication will fail. If this value is not
set, the broker will accept a token that has no \"aud\" claim, which is
strongly discouraged.";
public static final String SASL_OAUTHBEARER_EXPECTED_ISSUER =
"sasl.oauthbearer.expected.issuer";
- public static final String SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC = "The
(optional) setting for the broker to use to verify that the JWT was created by
the expected issuer. The JWT will"
+ public static final String SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC = "The
(strongly recommended) setting for the broker to use to verify that the JWT was
created by the expected issuer. The JWT will"
+ " be inspected for the standard OAuth \"iss\" claim and if this
value is set, the broker will match it exactly against what is in the JWT's
\"iss\" claim. If there is no"
- + " match, the broker will reject the JWT and authentication will
fail.";
+ + " match, the broker will reject the JWT and authentication will
fail. If this value is not set, the broker will accept a JWT bearing any (or
no) issuer, which is strongly discouraged.";
public static final String SASL_OAUTHBEARER_HEADER_URLENCODE =
"sasl.oauthbearer.header.urlencode";
public static final boolean DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE =
false;
@@ -409,8 +409,8 @@ public class SaslConfigs {
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS,
ConfigDef.Type.LONG,
DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS,
ConfigDef.Importance.LOW,
SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS,
ConfigDef.Type.LONG, DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS,
ConfigDef.Importance.LOW, SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS,
ConfigDef.Type.INT, DEFAULT_SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS,
ConfigDef.Importance.LOW, SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
ConfigDef.Type.LIST, List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true,
false), ConfigDef.Importance.LOW, SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER,
ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW,
SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
ConfigDef.Type.LIST, List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true,
false), ConfigDef.Importance.HIGH, SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER,
ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH,
SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE,
ConfigDef.Type.BOOLEAN, DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE,
ConfigDef.Importance.LOW, SASL_OAUTHBEARER_HEADER_URLENCODE_DOC);
}
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
index c5289676439..704cc388ccd 100644
---
a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
+++
b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
@@ -28,6 +28,7 @@ import org.apache.kafka.common.utils.Utils;
import java.util.Collections;
import java.util.List;
+import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
@@ -218,6 +219,6 @@ public class BrokerSecurityConfigs {
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS, LONG,
SaslConfigs.DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS, LOW,
SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS, LONG,
SaslConfigs.DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS, LOW,
SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS, INT,
SaslConfigs.DEFAULT_SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS, LOW,
SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE, LIST,
List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true, false), LOW,
SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, STRING,
null, LOW, SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC);
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE, LIST,
List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true, false), HIGH,
SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, STRING,
null, HIGH, SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC);
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
index c69db033052..f313935f300 100644
---
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
+++
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
@@ -121,11 +121,23 @@ public class BrokerJwtValidator implements JwtValidator {
if (clockSkew != null)
jwtConsumerBuilder.setAllowedClockSkewInSeconds(clockSkew);
- if (!expectedAudiences.isEmpty())
+ if (!expectedAudiences.isEmpty()) {
jwtConsumerBuilder.setExpectedAudience(expectedAudiences.toArray(new
String[0]));
+ } else {
+ log.warn("The OAuth broker validator is configured with a JWKS
endpoint but without \"{}\", so it will accept" +
+ " a JWT that does not carry an \"aud\" (audience) claim. This
is strongly discouraged; set \"{}\" to the" +
+ " audience(s) your brokers expect so that the token audience
is verified.",
+ SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+ }
- if (expectedIssuer != null)
+ if (expectedIssuer != null) {
jwtConsumerBuilder.setExpectedIssuer(expectedIssuer);
+ } else {
+ log.warn("The OAuth broker validator is configured with a JWKS
endpoint but without \"{}\", so it will accept" +
+ " a JWT bearing any (or no) \"iss\" (issuer) claim. This is
strongly discouraged; set \"{}\" to the issuer" +
+ " URL of your OAuth/OIDC provider so that the token issuer is
verified.",
+ SASL_OAUTHBEARER_EXPECTED_ISSUER,
SASL_OAUTHBEARER_EXPECTED_ISSUER);
+ }
this.jwtConsumer = jwtConsumerBuilder
.setJwsAlgorithmConstraints(DISALLOW_NONE)
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
index 5f76f508513..b732846ddb2 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
@@ -20,19 +20,30 @@ package org.apache.kafka.common.security.oauthbearer;
import org.apache.kafka.common.config.SaslConfigs;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenBuilder;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.CloseableVerificationKeyResolver;
+import org.apache.kafka.common.utils.LogCaptureAppender;
+import org.apache.logging.log4j.Level;
import org.jose4j.jwk.PublicJsonWebKey;
import org.jose4j.jws.AlgorithmIdentifiers;
import org.jose4j.lang.InvalidAlgorithmException;
import org.junit.jupiter.api.Test;
+import java.util.List;
import java.util.Map;
+import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
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;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class BrokerJwtValidatorTest extends JwtValidatorTest {
+ private static final String ATTACKER_ISSUER =
"https://evil.example/attacker";;
+
+ private static final String EXPECTED_ISSUER = "https://idp.legit.example/";;
+
@Override
protected JwtValidator createJwtValidator(AccessTokenBuilder builder) {
CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
builder.jwk().getKey();
@@ -69,10 +80,13 @@ public class BrokerJwtValidatorTest extends
JwtValidatorTest {
.jwk(jwk)
.alg(AlgorithmIdentifiers.RSA_USING_SHA256)
.addCustomClaim(subClaimName, subject)
+ .addCustomClaim("iss", EXPECTED_ISSUER)
.subjectClaimName(subClaimName)
.subject(null);
JwtValidator validator = createJwtValidator(tokenBuilder);
- Map<String, ?> saslConfigs =
getSaslConfigs(SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME, subClaimName);
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME, subClaimName,
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER));
validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
// Validation should succeed (e.g. signature verification) even if sub
claim is missing
@@ -81,10 +95,63 @@ public class BrokerJwtValidatorTest extends
JwtValidatorTest {
assertEquals(subject, token.principalName());
}
+ @Test
+ public void testExpectedIssuerMismatch() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", ATTACKER_ISSUER);
+ String accessToken = builder.build();
+
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ Map<String, ?> saslConfigs = getSaslConfigs(
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER);
+ validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ JwtValidatorException e = assertThrows(JwtValidatorException.class,
+ () -> validator.validate(accessToken));
+ assertErrorMessageContains(e.getMessage(), ATTACKER_ISSUER);
+ }
+
+ @Test
+ public void testWarnsWhenExpectedIssuerUnset() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+
+ try (LogCaptureAppender appender =
LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(BrokerJwtValidator.class, Level.WARN);
+ validator.configure(getSaslConfigs(), OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ List<String> warnings = appender.getMessages("WARN");
+ assertTrue(warnings.stream().anyMatch(message ->
message.contains(SASL_OAUTHBEARER_EXPECTED_ISSUER)),
+ "Expected a WARN log mentioning " +
SASL_OAUTHBEARER_EXPECTED_ISSUER + ", but got: " + warnings);
+ }
+ }
+
+ @Test
+ public void testWarnsWhenExpectedAudienceUnset() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+
+ try (LogCaptureAppender appender =
LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(BrokerJwtValidator.class, Level.WARN);
+ validator.configure(getSaslConfigs(), OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ List<String> warnings = appender.getMessages("WARN");
+ assertTrue(warnings.stream().anyMatch(message ->
message.contains(SASL_OAUTHBEARER_EXPECTED_AUDIENCE)),
+ "Expected a WARN log mentioning " +
SASL_OAUTHBEARER_EXPECTED_AUDIENCE + ", but got: " + warnings);
+ }
+ }
+
private void testEncryptionAlgorithm(PublicJsonWebKey jwk, String alg)
throws Exception {
- AccessTokenBuilder builder = new
AccessTokenBuilder().jwk(jwk).alg(alg);
+ AccessTokenBuilder builder = new
AccessTokenBuilder().jwk(jwk).alg(alg).addCustomClaim("iss", EXPECTED_ISSUER);
JwtValidator validator = createJwtValidator(builder);
- validator.configure(getSaslConfigs(), OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+ validator.configure(getSaslConfigs(SASL_OAUTHBEARER_EXPECTED_ISSUER,
EXPECTED_ISSUER),
+ OAUTHBEARER_MECHANISM, getJaasConfigEntries());
String accessToken = builder.build();
OAuthBearerToken token = validator.validate(accessToken);
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
index cf3754a77ac..96fe75f2d3c 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
@@ -30,6 +30,7 @@ import org.jose4j.jws.AlgorithmIdentifiers;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
+import java.util.HashMap;
import java.util.Map;
import static
org.apache.kafka.common.config.internals.BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
@@ -40,6 +41,8 @@ import static
org.junit.jupiter.api.Assertions.assertInstanceOf;
public class DefaultJwtValidatorTest extends OAuthBearerTest {
+ private static final String EXPECTED_ISSUER = "https://idp.legit.example/";;
+
@AfterEach
public void tearDown() {
System.clearProperty(BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
@@ -69,14 +72,18 @@ public class DefaultJwtValidatorTest extends
OAuthBearerTest {
PublicJsonWebKey jwk = createRsaJwk();
AccessTokenBuilder builder = new AccessTokenBuilder()
.jwk(jwk)
- .alg(AlgorithmIdentifiers.RSA_USING_SHA256);
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", EXPECTED_ISSUER);
String accessToken = builder.build();
JsonWebKeySet jwks = new JsonWebKeySet(jwk);
String jwksJson =
jwks.toJson(JsonWebKey.OutputControlLevel.PUBLIC_ONLY);
String fileUrl = tempFile(jwksJson).toURI().toString();
System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, fileUrl);
- Map<String, ?> configs =
getSaslConfigs(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL, fileUrl);
+ Map<String, Object> rawConfigs = new HashMap<>();
+ rawConfigs.put(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL,
fileUrl);
+ rawConfigs.put(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER,
EXPECTED_ISSUER);
+ Map<String, ?> configs = getSaslConfigs(rawConfigs);
DefaultJwtValidator jwtValidator = new DefaultJwtValidator();
assertDoesNotThrow(() -> jwtValidator.configure(configs,
OAUTHBEARER_MECHANISM, getJaasConfigEntries()));
diff --git
a/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
b/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
index d1b2161f7cb..c7e26c1be7d 100644
--- a/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
@@ -95,6 +95,7 @@ class ClientOAuthIntegrationTest extends
IntegrationTestHarness with SaslSetup {
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_JAAS_CONFIG}",
s"${classOf[OAuthBearerLoginModule].getName} required ;")
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE}",
issuerId)
+
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER}",
mockOAuthServer.issuerUrl(issuerId).toString)
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL}",
jwksUrl)
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${BrokerSecurityConfigs.SASL_SERVER_CALLBACK_HANDLER_CLASS_CONFIG}",
classOf[OAuthBearerValidatorCallbackHandler].getName)
