You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/11/03 13:30:13 UTC
[GitHub] [kafka] junrao commented on a change in pull request #7751: KAFKA-7987: Reinitialize ZookeeperClient after auth failures
junrao commented on a change in pull request #7751:
URL: https://github.com/apache/kafka/pull/7751#discussion_r516157977
##########
File path: core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala
##########
@@ -81,7 +85,8 @@ class ZooKeeperClient(connectString: String,
private val zNodeChildChangeHandlers = new ConcurrentHashMap[String, ZNodeChildChangeHandler]().asScala
private val inFlightRequests = new Semaphore(maxInFlightRequests)
private val stateChangeHandlers = new ConcurrentHashMap[String, StateChangeHandler]().asScala
- private[zookeeper] val expiryScheduler = new KafkaScheduler(threads = 1, "zk-session-expiry-handler")
+ private[zookeeper] val reinitializeScheduler = new KafkaScheduler(threads = 1, "zk-client-reinit-")
Review comment:
This is an existing issue. Should we include ZooKeeperClient.name in the thread name to distinguish between different instances? If so, we probably need to replace all the spaces in name when using it as the thread name prefix.
##########
File path: core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerWithZkSaslTest.scala
##########
@@ -0,0 +1,197 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.security.authorizer
+
+import java.net.InetAddress
+import java.util
+import java.util.UUID
+import java.util.concurrent.{Executors, TimeUnit}
+
+import javax.security.auth.Subject
+import javax.security.auth.callback.CallbackHandler
+import kafka.api.SaslSetup
+import kafka.security.authorizer.AclEntry.WildcardHost
+import kafka.server.KafkaConfig
+import kafka.utils.JaasTestUtils.{JaasModule, JaasSection}
+import kafka.utils.{JaasTestUtils, TestUtils}
+import kafka.zk.{KafkaZkClient, ZooKeeperTestHarness}
+import kafka.zookeeper.ZooKeeperClient
+import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter, AclBinding, AclBindingFilter}
+import org.apache.kafka.common.acl.AclOperation.{READ, WRITE}
+import org.apache.kafka.common.acl.AclPermissionType.ALLOW
+import org.apache.kafka.common.network.{ClientInformation, ListenerName}
+import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.common.requests.{RequestContext, RequestHeader}
+import org.apache.kafka.common.resource.PatternType.LITERAL
+import org.apache.kafka.common.resource.ResourcePattern
+import org.apache.kafka.common.resource.ResourceType.TOPIC
+import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
+import org.apache.kafka.test.{TestUtils => JTestUtils}
+import org.apache.zookeeper.server.auth.DigestLoginModule
+import org.junit.Assert.assertEquals
+import org.junit.{After, Before, Test}
+
+import scala.jdk.CollectionConverters._
+import scala.collection.Seq
+
+class AclAuthorizerWithZkSaslTest extends ZooKeeperTestHarness with SaslSetup {
+
+ private val aclAuthorizer = new AclAuthorizer
+ private val aclAuthorizer2 = new AclAuthorizer
+ private val resource: ResourcePattern = new ResourcePattern(TOPIC, "foo-" + UUID.randomUUID(), LITERAL)
+ private val username = "alice"
+ private val principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, username)
+ private val requestContext = newRequestContext(principal, InetAddress.getByName("192.168.0.1"))
+ private val executor = Executors.newSingleThreadScheduledExecutor
+ private var config: KafkaConfig = _
+
+ @Before
+ override def setUp(): Unit = {
+ // Allow failed clients to avoid server closing the connection before reporting AuthFailed.
+ System.setProperty("zookeeper.allowSaslFailedClients", "true")
+
+ // Configure ZK SASL with TestableDigestLoginModule for clients to inject failures
+ TestableDigestLoginModule.reset()
+ val jaasSections = JaasTestUtils.zkSections
+ val serverJaas = jaasSections.filter(_.contextName == "Server")
+ val clientJaas = jaasSections.filter(_.contextName == "Client")
+ .map(section => new TestableJaasSection(section.contextName, section.modules))
+ startSasl(serverJaas ++ clientJaas)
+
+ // Increase maxUpdateRetries to avoid transient failures
+ aclAuthorizer.maxUpdateRetries = Int.MaxValue
+ aclAuthorizer2.maxUpdateRetries = Int.MaxValue
+
+ super.setUp()
+ config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(0, zkConnect))
+
+ aclAuthorizer.configure(config.originals)
+ aclAuthorizer2.configure(config.originals)
+ }
+
+ @After
+ override def tearDown(): Unit = {
+ executor.shutdownNow()
+ aclAuthorizer.close()
+ aclAuthorizer2.close()
+ super.tearDown()
+ TestableDigestLoginModule.reset()
+ }
+
+ def jaasSections: collection.Seq[JaasTestUtils.JaasSection] = {
+ val jaasSections = JaasTestUtils.zkSections
+ val serverJaas = jaasSections.filter(_.contextName == "Server")
+ val clientJaas = jaasSections.filter(_.contextName == "Client")
+ .map(section => new TestableJaasSection(section.contextName, section.modules))
+ serverJaas ++ clientJaas
+ }
+
+ @Test
+ def testAclUpdateWithSessionExpiration(): Unit = {
+ zkClient(aclAuthorizer).currentZooKeeper.getTestable.injectSessionExpiration()
+ zkClient(aclAuthorizer2).currentZooKeeper.getTestable.injectSessionExpiration()
+ verifyAclUpdate()
+ }
+
+ @Test
+ def testAclUpdateWithAuthFailureInUpdater(): Unit = {
+ injectTransientAuthenticationFailure(aclAuthorizer)
+ verifyAclUpdate()
+ }
+
+ @Test
+ def testAclUpdateWithAuthFailureInObserver(): Unit = {
Review comment:
Hmm, what additional cases does aclAuthorizer2 cover?
##########
File path: core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerWithZkSaslTest.scala
##########
@@ -0,0 +1,197 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.security.authorizer
+
+import java.net.InetAddress
+import java.util
+import java.util.UUID
+import java.util.concurrent.{Executors, TimeUnit}
+
+import javax.security.auth.Subject
+import javax.security.auth.callback.CallbackHandler
+import kafka.api.SaslSetup
+import kafka.security.authorizer.AclEntry.WildcardHost
+import kafka.server.KafkaConfig
+import kafka.utils.JaasTestUtils.{JaasModule, JaasSection}
+import kafka.utils.{JaasTestUtils, TestUtils}
+import kafka.zk.{KafkaZkClient, ZooKeeperTestHarness}
+import kafka.zookeeper.ZooKeeperClient
+import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter, AclBinding, AclBindingFilter}
+import org.apache.kafka.common.acl.AclOperation.{READ, WRITE}
+import org.apache.kafka.common.acl.AclPermissionType.ALLOW
+import org.apache.kafka.common.network.{ClientInformation, ListenerName}
+import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.common.requests.{RequestContext, RequestHeader}
+import org.apache.kafka.common.resource.PatternType.LITERAL
+import org.apache.kafka.common.resource.ResourcePattern
+import org.apache.kafka.common.resource.ResourceType.TOPIC
+import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
+import org.apache.kafka.test.{TestUtils => JTestUtils}
+import org.apache.zookeeper.server.auth.DigestLoginModule
+import org.junit.Assert.assertEquals
+import org.junit.{After, Before, Test}
+
+import scala.jdk.CollectionConverters._
+import scala.collection.Seq
+
+class AclAuthorizerWithZkSaslTest extends ZooKeeperTestHarness with SaslSetup {
+
+ private val aclAuthorizer = new AclAuthorizer
+ private val aclAuthorizer2 = new AclAuthorizer
+ private val resource: ResourcePattern = new ResourcePattern(TOPIC, "foo-" + UUID.randomUUID(), LITERAL)
+ private val username = "alice"
+ private val principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, username)
+ private val requestContext = newRequestContext(principal, InetAddress.getByName("192.168.0.1"))
+ private val executor = Executors.newSingleThreadScheduledExecutor
+ private var config: KafkaConfig = _
+
+ @Before
+ override def setUp(): Unit = {
+ // Allow failed clients to avoid server closing the connection before reporting AuthFailed.
+ System.setProperty("zookeeper.allowSaslFailedClients", "true")
+
+ // Configure ZK SASL with TestableDigestLoginModule for clients to inject failures
+ TestableDigestLoginModule.reset()
+ val jaasSections = JaasTestUtils.zkSections
+ val serverJaas = jaasSections.filter(_.contextName == "Server")
+ val clientJaas = jaasSections.filter(_.contextName == "Client")
+ .map(section => new TestableJaasSection(section.contextName, section.modules))
+ startSasl(serverJaas ++ clientJaas)
+
+ // Increase maxUpdateRetries to avoid transient failures
+ aclAuthorizer.maxUpdateRetries = Int.MaxValue
+ aclAuthorizer2.maxUpdateRetries = Int.MaxValue
+
+ super.setUp()
+ config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(0, zkConnect))
+
+ aclAuthorizer.configure(config.originals)
+ aclAuthorizer2.configure(config.originals)
+ }
+
+ @After
+ override def tearDown(): Unit = {
+ executor.shutdownNow()
+ aclAuthorizer.close()
+ aclAuthorizer2.close()
+ super.tearDown()
+ TestableDigestLoginModule.reset()
+ }
+
+ def jaasSections: collection.Seq[JaasTestUtils.JaasSection] = {
+ val jaasSections = JaasTestUtils.zkSections
+ val serverJaas = jaasSections.filter(_.contextName == "Server")
+ val clientJaas = jaasSections.filter(_.contextName == "Client")
+ .map(section => new TestableJaasSection(section.contextName, section.modules))
+ serverJaas ++ clientJaas
+ }
+
+ @Test
+ def testAclUpdateWithSessionExpiration(): Unit = {
+ zkClient(aclAuthorizer).currentZooKeeper.getTestable.injectSessionExpiration()
+ zkClient(aclAuthorizer2).currentZooKeeper.getTestable.injectSessionExpiration()
+ verifyAclUpdate()
+ }
+
+ @Test
+ def testAclUpdateWithAuthFailureInUpdater(): Unit = {
+ injectTransientAuthenticationFailure(aclAuthorizer)
+ verifyAclUpdate()
+ }
+
+ @Test
+ def testAclUpdateWithAuthFailureInObserver(): Unit = {
+ injectTransientAuthenticationFailure(aclAuthorizer2)
+ verifyAclUpdate()
+ }
+
+ private def injectTransientAuthenticationFailure(authorizer: AclAuthorizer): Unit = {
+ TestableDigestLoginModule.injectInvalidCredentials()
+ zkClient(authorizer).currentZooKeeper.getTestable.injectSessionExpiration()
+ executor.schedule((() => TestableDigestLoginModule.reset()): Runnable,
+ ZooKeeperClient.AuthFailedRetryBackoffMs * 2, TimeUnit.MILLISECONDS)
+ }
+
+ private def verifyAclUpdate(): Unit = {
+ val allowReadAcl = new AccessControlEntry(principal.toString, WildcardHost, READ, ALLOW)
+ val allowWriteAcl = new AccessControlEntry(principal.toString, WildcardHost, WRITE, ALLOW)
+ val acls = Set(allowReadAcl, allowWriteAcl)
+
+ TestUtils.retry(maxWaitMs = 15000) {
+ try {
+ addAcls(aclAuthorizer, acls, resource)
+ } catch {
+ case e: Exception => // Ignore error and retry
+ }
+ assertEquals(acls, getAcls(aclAuthorizer, resource))
+ }
+ val (acls2, _) = TestUtils.computeUntilTrue(getAcls(aclAuthorizer2, resource)) { _ == acls }
+ assertEquals(acls, acls2)
+ }
+
+ private def zkClient(authorizer: AclAuthorizer): KafkaZkClient = {
+ JTestUtils.fieldValue(authorizer, classOf[AclAuthorizer], "zkClient")
+ }
+
+ private def addAcls(authorizer: AclAuthorizer, aces: Set[AccessControlEntry], resourcePattern: ResourcePattern): Unit = {
+ val bindings = aces.map { ace => new AclBinding(resourcePattern, ace) }
+ authorizer.createAcls(requestContext, bindings.toList.asJava).asScala
+ .map(_.toCompletableFuture.get)
+ .foreach { result => result.exception.ifPresent { e => throw e } }
+ }
+
+ private def getAcls(authorizer: AclAuthorizer, resourcePattern: ResourcePattern): Set[AccessControlEntry] = {
+ val acls = authorizer.acls(new AclBindingFilter(resourcePattern.toFilter, AccessControlEntryFilter.ANY)).asScala.toSet
+ acls.map(_.entry)
+ }
+
+ private def newRequestContext(principal: KafkaPrincipal, clientAddress: InetAddress, apiKey: ApiKeys = ApiKeys.PRODUCE): RequestContext = {
+ val securityProtocol = SecurityProtocol.SASL_PLAINTEXT
+ val header = new RequestHeader(apiKey, 2, "", 1) //ApiKeys apiKey, short version, String clientId, int correlation
+ new RequestContext(header, "", clientAddress, principal, ListenerName.forSecurityProtocol(securityProtocol),
+ securityProtocol, ClientInformation.EMPTY)
+ }
+}
+
+object TestableDigestLoginModule {
+ var injectedPassword: Option[String] = None
Review comment:
Does injectedPassword need to be volatile?
##########
File path: core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala
##########
@@ -39,6 +39,10 @@ import scala.jdk.CollectionConverters._
import scala.collection.Seq
import scala.collection.mutable.Set
+object ZooKeeperClient {
+ val AuthFailedRetryBackoffMs = 100
Review comment:
In reinitialize(), we back off by 1 sec if the creation of ZooKeeper instance fails. Perhaps we should make this 1 sec too to be more consistent?
##########
File path: core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerWithZkSaslTest.scala
##########
@@ -0,0 +1,197 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.security.authorizer
+
+import java.net.InetAddress
+import java.util
+import java.util.UUID
+import java.util.concurrent.{Executors, TimeUnit}
+
+import javax.security.auth.Subject
+import javax.security.auth.callback.CallbackHandler
+import kafka.api.SaslSetup
+import kafka.security.authorizer.AclEntry.WildcardHost
+import kafka.server.KafkaConfig
+import kafka.utils.JaasTestUtils.{JaasModule, JaasSection}
+import kafka.utils.{JaasTestUtils, TestUtils}
+import kafka.zk.{KafkaZkClient, ZooKeeperTestHarness}
+import kafka.zookeeper.ZooKeeperClient
+import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter, AclBinding, AclBindingFilter}
+import org.apache.kafka.common.acl.AclOperation.{READ, WRITE}
+import org.apache.kafka.common.acl.AclPermissionType.ALLOW
+import org.apache.kafka.common.network.{ClientInformation, ListenerName}
+import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.common.requests.{RequestContext, RequestHeader}
+import org.apache.kafka.common.resource.PatternType.LITERAL
+import org.apache.kafka.common.resource.ResourcePattern
+import org.apache.kafka.common.resource.ResourceType.TOPIC
+import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
+import org.apache.kafka.test.{TestUtils => JTestUtils}
+import org.apache.zookeeper.server.auth.DigestLoginModule
+import org.junit.Assert.assertEquals
+import org.junit.{After, Before, Test}
+
+import scala.jdk.CollectionConverters._
+import scala.collection.Seq
+
+class AclAuthorizerWithZkSaslTest extends ZooKeeperTestHarness with SaslSetup {
+
+ private val aclAuthorizer = new AclAuthorizer
+ private val aclAuthorizer2 = new AclAuthorizer
+ private val resource: ResourcePattern = new ResourcePattern(TOPIC, "foo-" + UUID.randomUUID(), LITERAL)
+ private val username = "alice"
+ private val principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, username)
+ private val requestContext = newRequestContext(principal, InetAddress.getByName("192.168.0.1"))
+ private val executor = Executors.newSingleThreadScheduledExecutor
+ private var config: KafkaConfig = _
+
+ @Before
+ override def setUp(): Unit = {
+ // Allow failed clients to avoid server closing the connection before reporting AuthFailed.
+ System.setProperty("zookeeper.allowSaslFailedClients", "true")
+
+ // Configure ZK SASL with TestableDigestLoginModule for clients to inject failures
+ TestableDigestLoginModule.reset()
+ val jaasSections = JaasTestUtils.zkSections
+ val serverJaas = jaasSections.filter(_.contextName == "Server")
+ val clientJaas = jaasSections.filter(_.contextName == "Client")
+ .map(section => new TestableJaasSection(section.contextName, section.modules))
+ startSasl(serverJaas ++ clientJaas)
+
+ // Increase maxUpdateRetries to avoid transient failures
+ aclAuthorizer.maxUpdateRetries = Int.MaxValue
+ aclAuthorizer2.maxUpdateRetries = Int.MaxValue
+
+ super.setUp()
+ config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(0, zkConnect))
+
+ aclAuthorizer.configure(config.originals)
+ aclAuthorizer2.configure(config.originals)
+ }
+
+ @After
+ override def tearDown(): Unit = {
+ executor.shutdownNow()
+ aclAuthorizer.close()
+ aclAuthorizer2.close()
+ super.tearDown()
+ TestableDigestLoginModule.reset()
+ }
+
+ def jaasSections: collection.Seq[JaasTestUtils.JaasSection] = {
+ val jaasSections = JaasTestUtils.zkSections
+ val serverJaas = jaasSections.filter(_.contextName == "Server")
+ val clientJaas = jaasSections.filter(_.contextName == "Client")
+ .map(section => new TestableJaasSection(section.contextName, section.modules))
+ serverJaas ++ clientJaas
+ }
+
+ @Test
+ def testAclUpdateWithSessionExpiration(): Unit = {
+ zkClient(aclAuthorizer).currentZooKeeper.getTestable.injectSessionExpiration()
+ zkClient(aclAuthorizer2).currentZooKeeper.getTestable.injectSessionExpiration()
+ verifyAclUpdate()
+ }
+
+ @Test
+ def testAclUpdateWithAuthFailureInUpdater(): Unit = {
+ injectTransientAuthenticationFailure(aclAuthorizer)
+ verifyAclUpdate()
+ }
+
+ @Test
+ def testAclUpdateWithAuthFailureInObserver(): Unit = {
+ injectTransientAuthenticationFailure(aclAuthorizer2)
+ verifyAclUpdate()
+ }
+
+ private def injectTransientAuthenticationFailure(authorizer: AclAuthorizer): Unit = {
+ TestableDigestLoginModule.injectInvalidCredentials()
+ zkClient(authorizer).currentZooKeeper.getTestable.injectSessionExpiration()
+ executor.schedule((() => TestableDigestLoginModule.reset()): Runnable,
+ ZooKeeperClient.AuthFailedRetryBackoffMs * 2, TimeUnit.MILLISECONDS)
+ }
+
+ private def verifyAclUpdate(): Unit = {
+ val allowReadAcl = new AccessControlEntry(principal.toString, WildcardHost, READ, ALLOW)
+ val allowWriteAcl = new AccessControlEntry(principal.toString, WildcardHost, WRITE, ALLOW)
+ val acls = Set(allowReadAcl, allowWriteAcl)
+
+ TestUtils.retry(maxWaitMs = 15000) {
+ try {
+ addAcls(aclAuthorizer, acls, resource)
+ } catch {
+ case e: Exception => // Ignore error and retry
Review comment:
e is unused.
##########
File path: core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerWithZkSaslTest.scala
##########
@@ -0,0 +1,197 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.security.authorizer
+
+import java.net.InetAddress
+import java.util
+import java.util.UUID
+import java.util.concurrent.{Executors, TimeUnit}
+
+import javax.security.auth.Subject
+import javax.security.auth.callback.CallbackHandler
+import kafka.api.SaslSetup
+import kafka.security.authorizer.AclEntry.WildcardHost
+import kafka.server.KafkaConfig
+import kafka.utils.JaasTestUtils.{JaasModule, JaasSection}
+import kafka.utils.{JaasTestUtils, TestUtils}
+import kafka.zk.{KafkaZkClient, ZooKeeperTestHarness}
+import kafka.zookeeper.ZooKeeperClient
+import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter, AclBinding, AclBindingFilter}
+import org.apache.kafka.common.acl.AclOperation.{READ, WRITE}
+import org.apache.kafka.common.acl.AclPermissionType.ALLOW
+import org.apache.kafka.common.network.{ClientInformation, ListenerName}
+import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.common.requests.{RequestContext, RequestHeader}
+import org.apache.kafka.common.resource.PatternType.LITERAL
+import org.apache.kafka.common.resource.ResourcePattern
+import org.apache.kafka.common.resource.ResourceType.TOPIC
+import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
+import org.apache.kafka.test.{TestUtils => JTestUtils}
+import org.apache.zookeeper.server.auth.DigestLoginModule
+import org.junit.Assert.assertEquals
+import org.junit.{After, Before, Test}
+
+import scala.jdk.CollectionConverters._
+import scala.collection.Seq
+
+class AclAuthorizerWithZkSaslTest extends ZooKeeperTestHarness with SaslSetup {
+
+ private val aclAuthorizer = new AclAuthorizer
+ private val aclAuthorizer2 = new AclAuthorizer
+ private val resource: ResourcePattern = new ResourcePattern(TOPIC, "foo-" + UUID.randomUUID(), LITERAL)
+ private val username = "alice"
+ private val principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, username)
+ private val requestContext = newRequestContext(principal, InetAddress.getByName("192.168.0.1"))
+ private val executor = Executors.newSingleThreadScheduledExecutor
+ private var config: KafkaConfig = _
+
+ @Before
+ override def setUp(): Unit = {
+ // Allow failed clients to avoid server closing the connection before reporting AuthFailed.
+ System.setProperty("zookeeper.allowSaslFailedClients", "true")
+
+ // Configure ZK SASL with TestableDigestLoginModule for clients to inject failures
+ TestableDigestLoginModule.reset()
+ val jaasSections = JaasTestUtils.zkSections
+ val serverJaas = jaasSections.filter(_.contextName == "Server")
+ val clientJaas = jaasSections.filter(_.contextName == "Client")
+ .map(section => new TestableJaasSection(section.contextName, section.modules))
+ startSasl(serverJaas ++ clientJaas)
+
+ // Increase maxUpdateRetries to avoid transient failures
+ aclAuthorizer.maxUpdateRetries = Int.MaxValue
+ aclAuthorizer2.maxUpdateRetries = Int.MaxValue
+
+ super.setUp()
+ config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(0, zkConnect))
+
+ aclAuthorizer.configure(config.originals)
+ aclAuthorizer2.configure(config.originals)
+ }
+
+ @After
+ override def tearDown(): Unit = {
+ executor.shutdownNow()
+ aclAuthorizer.close()
+ aclAuthorizer2.close()
+ super.tearDown()
+ TestableDigestLoginModule.reset()
+ }
+
+ def jaasSections: collection.Seq[JaasTestUtils.JaasSection] = {
Review comment:
jaasSections seems unused?
##########
File path: core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala
##########
@@ -39,6 +39,10 @@ import scala.jdk.CollectionConverters._
import scala.collection.Seq
import scala.collection.mutable.Set
+object ZooKeeperClient {
+ val AuthFailedRetryBackoffMs = 1000
Review comment:
Perhaps we can make this constant more general and reuse it in reinitialize().
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org