[kafka] branch trunk updated: KAFKA-13443: Kafka broker exits when OAuth enabled and certain configuration not specified (#11484)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dajac 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 a448ddb  KAFKA-13443: Kafka broker exits when OAuth enabled and certain configuration not specified (#11484)
a448ddb is described below

commit a448ddbecb5036baafea039325e644e8bf866507
Author: Kirk True <ki...@mustardgrain.com>
AuthorDate: Wed Nov 17 02:17:26 2021 -0800

    KAFKA-13443: Kafka broker exits when OAuth enabled and certain configuration not specified (#11484)
    
    The sasl.oauthbearer.jwks.endpoint.retry.backoff.ms and sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms configuration options were added to the SaslConfig class but their default values were not added to KafkaConfig. As a result, when the OAuth validation feature is enabled in the broker and those two configuration values aren't explicitly provided by the user, the broker exits. This patch fixes the issue by defining them in the KafkaConfig class.
    
    Reviewers: David Jacot <dj...@confluent.io>
---
 core/src/main/scala/kafka/server/KafkaConfig.scala          | 2 ++
 core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala | 9 +++++++++
 2 files changed, 11 insertions(+)

diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 33221c6..9b67acd 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -1342,6 +1342,8 @@ object KafkaConfig {
       .define(SaslOAuthBearerTokenEndpointUrlProp, STRING, null, MEDIUM, SaslOAuthBearerTokenEndpointUrlDoc)
       .define(SaslOAuthBearerJwksEndpointUrlProp, STRING, null, MEDIUM, SaslOAuthBearerJwksEndpointUrlDoc)
       .define(SaslOAuthBearerJwksEndpointRefreshMsProp, LONG, Defaults.SaslOAuthBearerJwksEndpointRefreshMs, LOW, SaslOAuthBearerJwksEndpointRefreshMsDoc)
+      .define(SaslOAuthBearerJwksEndpointRetryBackoffMsProp, LONG, Defaults.SaslOAuthBearerJwksEndpointRetryBackoffMs, LOW, SaslOAuthBearerJwksEndpointRetryBackoffMsDoc)
+      .define(SaslOAuthBearerJwksEndpointRetryBackoffMaxMsProp, LONG, Defaults.SaslOAuthBearerJwksEndpointRetryBackoffMaxMs, LOW, SaslOAuthBearerJwksEndpointRetryBackoffMaxMsDoc)
       .define(SaslOAuthBearerClockSkewSecondsProp, INT, Defaults.SaslOAuthBearerClockSkewSeconds, LOW, SaslOAuthBearerClockSkewSecondsDoc)
       .define(SaslOAuthBearerExpectedAudienceProp, LIST, null, LOW, SaslOAuthBearerExpectedAudienceDoc)
       .define(SaslOAuthBearerExpectedIssuerProp, STRING, null, LOW, SaslOAuthBearerExpectedIssuerDoc)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 2b43806..05100df 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1321,4 +1321,13 @@ class KafkaConfigTest {
     assertEquals("3", originals.get(KafkaConfig.BrokerIdProp))
     assertEquals("3", originals.get(KafkaConfig.NodeIdProp))
   }
+
+  @Test
+  def testSaslJwksEndpointRetryDefaults(): Unit = {
+    val props = new Properties()
+    props.put(KafkaConfig.ZkConnectProp, "localhost:2181")
+    val config = KafkaConfig.fromProps(props)
+    assertNotNull(config.getLong(KafkaConfig.SaslOAuthBearerJwksEndpointRetryBackoffMsProp))
+    assertNotNull(config.getLong(KafkaConfig.SaslOAuthBearerJwksEndpointRetryBackoffMaxMsProp))
+  }
 }