Fix and reenable testChrootExistsAndRootIsLocked test
Reviewers: Manikumar Reddy Igor Soarez
*/
package kafka.zk
-import java.util.{Base64, Collections, Properties}
+import java.util.{Collections, Properties}
import java.nio.charset.StandardCharsets.UTF_8
-import java.security.MessageDigest
import java.util.concurrent.{CountDownLatch, TimeUnit}
-
import kafka.api.{ApiVersion, LeaderAndIsr}
import kafka.cluster.{Broker, EndPoint}
import kafka.log.LogConfig
@@ -31,9 +29,9 @@ import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.common.security.token.delegation.TokenInformation
import org.apache.kafka.common.utils.{SecurityUtils, Time}
-import org.apache.zookeeper.KeeperException.{Code, NoNodeException, NodeExistsException}
+import org.apache.zookeeper.KeeperException.{Code, NoAuthException, NoNodeException, NodeExistsException}
import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.{AfterEach, BeforeEach, Disabled, Test}
+import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import scala.jdk.CollectionConverters._
import scala.collection.mutable.ArrayBuffer
@@ -53,7 +51,7 @@ import org.apache.kafka.common.resource.ResourceType.{GROUP, TOPIC}
import org.apache.kafka.common.security.JaasUtils
import org.apache.zookeeper.ZooDefs
import org.apache.zookeeper.client.ZKClientConfig
-import org.apache.zookeeper.data.{ACL, Id, Stat}
+import org.apache.zookeeper.data.Stat
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource
@@ -142,24 +140,23 @@ class KafkaZkClientTest extends ZooKeeperTestHarness {
}
}
- @Disabled
@Test
def testChrootExistsAndRootIsLocked(): Unit = {
// chroot is accessible
val root = "/testChrootExistsAndRootIsLocked"
val chroot = s"$root/chroot"
+
zkClient.makeSurePersistentPathExists(chroot)
zkClient.setAcl(chroot, ZooDefs.Ids.OPEN_ACL_UNSAFE.asScala)
- // root is inaccessible
- val scheme = "digest"
- val id = "test"
- val pwd = "12345"
- val digest = Base64.getEncoder.encode(MessageDigest.getInstance("SHA1").digest(s"$id:$pwd".getBytes()))
- zkClient.currentZooKeeper.addAuthInfo(scheme, digest)
- zkClient.setAcl(root, Seq(new ACL(ZooDefs.Perms.ALL, new Id(scheme, s"$id:$digest"))))
+ // root is read-only
+ zkClient.setAcl(root, ZooDefs.Ids.READ_ACL_UNSAFE.asScala)
+
+ // we should not be able to create node under chroot folder
+ assertThrows(classOf[NoAuthException], () => zkClient.makeSurePersistentPathExists(chroot))
- // this client won't have access to the root, but the chroot already exists
+ // this client doesn't have create permission to the root and chroot, but the chroot already exists
+ // Expect that no exception thrown
val chrootClient = KafkaZkClient(zkConnect + chroot, zkAclsEnabled.getOrElse(JaasUtils.isZkSaslEnabled), zkSessionTimeout,
zkConnectionTimeout, zkMaxInFlightRequests, Time.SYSTEM, createChrootIfNecessary = true)
chrootClient.close()
