rajinisivaram commented on a change in pull request #7751:
URL: https://github.com/apache/kafka/pull/7751#discussion_r515933228
##########
File path: core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala
##########
@@ -437,8 +437,14 @@ class ZooKeeperClient(connectString: String,
if (state == KeeperState.AuthFailed) {
error("Auth failed.")
stateChangeHandlers.values.foreach(_.onAuthFailure())
+
+ // If this is during initial startup, the reinitialization
scheduler hasn't been started yet.
+ // To support failing fast, allow authorization to fail in that
case.
+ if (reinitializeScheduler.isStarted) {
Review comment:
Fixed.
##########
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:
Done.
##########
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:
Done.
##########
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:
Removed.
##########
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:
One writes to ZK, the other just uses the watcher. But there is no
reason for two tests, converted to one.
##########
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:
Removed.
##########
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:
Updated.
----------------------------------------------------------------
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:
us...@infra.apache.org
