[KYUUBI #1176] InvalidACL appears in the engine when zookeeper acl is turned on

### _Why are the changes needed?_
https://github.com/apache/incubator-kyuubi/issues/1176
When `kyuubi.ha.zookeeper.acl.enabled=true`, both service and engine will use zookeeper acl to create znode, but engine has no keytab information and cannot write information to zookeeper, throwing an exception.

```java
Caused by: org.apache.zookeeper.KeeperException$InvalidACLException: KeeperErrorCode = InvalidACL for /kyuubi_USER/XXXX
	at org.apache.zookeeper.KeeperException.create(KeeperException.java:124)
	at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)
	at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:792)
	at org.apache.kyuubi.shade.org.apache.curator.framework.imps.CreateBuilderImpl$11.call(CreateBuilderImpl.java:740)
```

### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [x] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.readthedocs.io/en/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #1177 from cxzl25/KYUUBI-1176.

Closes #1176

ecc08fa7 [sychen] fix engine acl
0b7cc2ec [sychen] fix InvalidACL

Authored-by: sychen <sychen@trip.com>
Signed-off-by: Kent Yao <yao@apache.org>
(cherry picked from commit d332534325)
Signed-off-by: Kent Yao <yao@apache.org>

docs/deployment/settings.md

@@ -192,6 +192,7 @@ kyuubi\.frontend<br>\.worker\.keepalive\.time|<div style='width: 65pt;word-wrap:
 Key | Default | Meaning | Type | Since
 --- | --- | --- | --- | ---
 kyuubi\.ha\.zookeeper<br>\.acl\.enabled|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>false</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Set to true if the zookeeper ensemble is kerberized</div>|<div style='width: 30pt'>boolean</div>|<div style='width: 20pt'>1.0.0</div>
+kyuubi\.ha\.zookeeper<br>\.acl\.engine\.enabled|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>false</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Set to true if the zookeeper ensemble is kerberized at engine side.</div>|<div style='width: 30pt'>boolean</div>|<div style='width: 20pt'>1.4.0</div>
 kyuubi\.ha\.zookeeper<br>\.connection\.base\.retry<br>\.wait|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>1000</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Initial amount of time to wait between retries to the zookeeper ensemble</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.0.0</div>
 kyuubi\.ha\.zookeeper<br>\.connection\.max<br>\.retries|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>3</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Max retry times for connecting to the zookeeper ensemble</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.0.0</div>
 kyuubi\.ha\.zookeeper<br>\.connection\.max\.retry<br>\.wait|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>30000</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Max amount of time to wait between retries for BOUNDED_EXPONENTIAL_BACKOFF policy can reach, or max time until elapsed for UNTIL_ELAPSED policy to connect the zookeeper ensemble</div>|<div style='width: 30pt'>int</div>|<div style='width: 20pt'>1.0.0</div>

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/HighAvailabilityConf.scala

@@ -48,6 +48,13 @@ object HighAvailabilityConf {
       .booleanConf
       .createWithDefault(UserGroupInformation.isSecurityEnabled)
 
+  val HA_ZK_ACL_ENGINE_ENABLED: ConfigEntry[Boolean] =
+    buildConf("ha.zookeeper.acl.engine.enabled")
+      .doc("Set to true if the zookeeper ensemble is kerberized at engine side.")
+      .version("1.4.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val HA_ZK_CONN_MAX_RETRIES: ConfigEntry[Int] =
     buildConf("ha.zookeeper.connection.max.retries")
       .doc("Max retry times for connecting to the zookeeper ensemble")

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/ZooKeeperACLProvider.scala

@@ -33,11 +33,20 @@ class ZooKeeperACLProvider(conf: KyuubiConf) extends ACLProvider {
    */
   override lazy val getDefaultAcl: java.util.List[ACL] = {
     val nodeAcls = new java.util.ArrayList[ACL]
-    if (conf.get(HighAvailabilityConf.HA_ZK_ACL_ENABLED)) {
+
+    def addACL(): Unit = {
       // Read all to the world
       nodeAcls.addAll(ZooDefs.Ids.READ_ACL_UNSAFE)
       // Create/Delete/Write/Admin to the authenticated user
       nodeAcls.addAll(ZooDefs.Ids.CREATOR_ALL_ACL)
+    }
+
+    if (conf.get(HighAvailabilityConf.HA_ZK_ACL_ENABLED) &&
+      conf.get(HighAvailabilityConf.HA_ZK_ENGINE_REF_ID).isEmpty) {
+      addACL()
+    } else if (conf.get(HighAvailabilityConf.HA_ZK_ACL_ENGINE_ENABLED) &&
+      conf.get(HighAvailabilityConf.HA_ZK_ENGINE_REF_ID).nonEmpty) {
+      addACL()
     } else {
       // ACLs for znodes on a non-kerberized cluster
       // Create/Read/Delete/Write/Admin to the world

kyuubi-ha/src/test/scala/org/apache/kyuubi/ha/client/ServiceDiscoverySuite.scala

@@ -19,12 +19,14 @@ package org.apache.kyuubi.ha.client
 
 import java.io.{File, IOException}
 import java.net.InetAddress
+import java.util
 import javax.security.auth.login.Configuration
 
 import scala.collection.JavaConverters._
 
 import org.apache.hadoop.util.StringUtils
 import org.apache.zookeeper.ZooDefs
+import org.apache.zookeeper.data.ACL
 import org.scalatest.time.SpanSugar._
 
 import org.apache.kyuubi.{KerberizedTestHelper, KYUUBI_VERSION}
@@ -97,17 +99,31 @@ class ServiceDiscoverySuite extends KerberizedTestHelper {
   }
 
   test("acl for zookeeper") {
-    val provider = new ZooKeeperACLProvider(conf)
-    val acl = provider.getDefaultAcl
-    assert(acl.size() === 1)
-    assert(acl === ZooDefs.Ids.OPEN_ACL_UNSAFE)
+    val expectedNoACL = new util.ArrayList[ACL](ZooDefs.Ids.OPEN_ACL_UNSAFE)
+    val expectedEnableACL = new util.ArrayList[ACL](ZooDefs.Ids.READ_ACL_UNSAFE)
+    expectedEnableACL.addAll(ZooDefs.Ids.CREATOR_ALL_ACL)
 
-    val conf1 = conf.clone.set(HA_ZK_ACL_ENABLED, true)
-    val acl1 = new ZooKeeperACLProvider(conf1).getDefaultAcl
-    assert(acl1.size() === 2)
-    val expected = ZooDefs.Ids.READ_ACL_UNSAFE
-    expected.addAll(ZooDefs.Ids.CREATOR_ALL_ACL)
-    assert(acl1 === expected)
+    def assertACL(expected: util.List[ACL], actual: util.List[ACL]): Unit = {
+      assert(actual.size() == expected.size())
+      assert(actual === expected)
+    }
+
+    val acl = new ZooKeeperACLProvider(conf).getDefaultAcl
+    assertACL(expectedNoACL, acl)
+
+    val serverConf = conf.clone.set(HA_ZK_ACL_ENABLED, true)
+    val serverACL = new ZooKeeperACLProvider(serverConf).getDefaultAcl
+    assertACL(expectedEnableACL, serverACL)
+
+    val engineConf = serverConf.clone.set(HA_ZK_ENGINE_REF_ID, "ref")
+    engineConf.set(HA_ZK_ACL_ENGINE_ENABLED, false)
+    val engineACL = new ZooKeeperACLProvider(engineConf).getDefaultAcl
+    assertACL(expectedNoACL, engineACL)
+
+    val enableEngineACLConf = serverConf.clone.set(HA_ZK_ENGINE_REF_ID, "ref")
+    enableEngineACLConf.set(HA_ZK_ACL_ENGINE_ENABLED, true)
+    val enableEngineACL = new ZooKeeperACLProvider(enableEngineACLConf).getDefaultAcl
+    assertACL(expectedEnableACL, enableEngineACL)
   }
 
   test("set up zookeeper auth") {