KAFKA-7007: All ACL changes should use single /kafka-acl-changes path

core/src/main/scala/kafka/security/auth/Resource.scala

@@ -16,6 +16,7 @@
  */
 package kafka.security.auth
 
+import kafka.common.KafkaException
 import org.apache.kafka.common.resource.{ResourceNameType, ResourcePattern}
 
 object Resource {
@@ -26,16 +27,18 @@ object Resource {
   val WildCardResource = "*"
 
   def fromString(str: String): Resource = {
-    ResourceNameType.values.find(nameType => str.startsWith(nameType.name)) match {
-      case Some(nameType) =>
-        str.split(Separator, 3) match {
-          case Array(_, resourceType, name, _*) => new Resource(ResourceType.fromString(resourceType), name, nameType)
-          case _ => throw new IllegalArgumentException("expected a string in format ResourceType:ResourceName but got " + str)
-        }
-      case _ =>
-        str.split(Separator, 2) match {
-          case Array(resourceType, name, _*) => new Resource(ResourceType.fromString(resourceType), name, ResourceNameType.LITERAL)
-          case _ => throw new IllegalArgumentException("expected a string in format ResourceType:ResourceName but got " + str)
+    ResourceType.values.find(resourceType => str.startsWith(resourceType.name + Separator)) match {
+      case None => throw new KafkaException("Invalid resource string: '" + str + "'")
+      case Some(resourceType) =>
+        val remaining = str.substring(resourceType.name.length + 1)
+
+        ResourceNameType.values.find(nameType => remaining.startsWith(nameType.name + Separator)) match {
+          case Some(nameType) =>
+            val name = remaining.substring(nameType.name.length + 1)
+            Resource(resourceType, name, nameType)
+
+          case None =>
+            Resource(resourceType, remaining, ResourceNameType.LITERAL)
         }
     }
   }
@@ -74,7 +77,7 @@ case class Resource(resourceType: ResourceType, name: String, nameType: Resource
   }
 
   override def toString: String = {
-    nameType + Resource.Separator + resourceType.name + Resource.Separator + name
+    resourceType.name + Resource.Separator + nameType + Resource.Separator + name
   }
 }
 

core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala

@@ -20,13 +20,15 @@ import java.util
 import java.util.concurrent.locks.ReentrantReadWriteLock
 
 import com.typesafe.scalalogging.Logger
+import kafka.api.KAFKA_2_0_IV1
 import kafka.common.{NotificationHandler, ZkNodeChangeNotificationListener}
 import kafka.network.RequestChannel.Session
 import kafka.security.auth.SimpleAclAuthorizer.VersionedAcls
 import kafka.server.KafkaConfig
 import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
 import kafka.utils._
-import kafka.zk.{AclChangeNotificationSequenceZNode, KafkaZkClient, ZkAclStore}
+import kafka.zk.{AclChangeNotificationSequenceZNode, AclChangeNotificationZNode, KafkaZkClient, ZkAclStore}
+import org.apache.kafka.common.errors.UnsupportedVersionException
 import org.apache.kafka.common.resource.ResourceNameType
 import org.apache.kafka.common.security.auth.KafkaPrincipal
 import org.apache.kafka.common.utils.{SecurityUtils, Time}
@@ -55,7 +57,8 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
   private var superUsers = Set.empty[KafkaPrincipal]
   private var shouldAllowEveryoneIfNoAclIsFound = false
   private var zkClient: KafkaZkClient = _
-  private var aclChangeListeners: Seq[ZkNodeChangeNotificationListener] = List()
+  private var aclChangeListener: ZkNodeChangeNotificationListener = _
+  private var legacyChangeEvent: Boolean = _
 
   @volatile
   private var aclCache = new scala.collection.immutable.TreeMap[Resource, VersionedAcls]()(ResourceOrdering)
@@ -96,9 +99,12 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
       zkMaxInFlightRequests, time, "kafka.security", "SimpleAclAuthorizer")
     zkClient.createAclPaths()
 
+    legacyChangeEvent = kafkaConfig.interBrokerProtocolVersion < KAFKA_2_0_IV1
+
     loadCache()
 
-    startZkChangeListeners()
+    aclChangeListener = new ZkNodeChangeNotificationListener(zkClient, AclChangeNotificationZNode.path, AclChangeNotificationSequenceZNode.SequenceNumberPrefix, AclChangedNotificationHandler)
+    aclChangeListener.init()
   }
 
   override def authorize(session: Session, operation: Operation, resource: Resource): Boolean = {
@@ -161,6 +167,11 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
 
   override def addAcls(acls: Set[Acl], resource: Resource) {
     if (acls != null && acls.nonEmpty) {
+      if (legacyChangeEvent && resource.nameType == ResourceNameType.PREFIXED) {
+        throw new UnsupportedVersionException(s"Adding ACLs on prefixed resource patterns requires " +
+          s"${KafkaConfig.InterBrokerProtocolVersionProp} of $KAFKA_2_0_IV1 or greater")
+      }
+
       inWriteLock(lock) {
         updateResourceAcls(resource) { currentAcls =>
           currentAcls ++ acls
@@ -231,7 +242,7 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
   }
 
   def close() {
-    aclChangeListeners.foreach(listener => listener.close())
+    if (aclChangeListener != null) aclChangeListener.close()
     if (zkClient != null) zkClient.close()
   }
 
@@ -251,16 +262,6 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
     }
   }
 
-  private def startZkChangeListeners(): Unit = {
-    aclChangeListeners = ZkAclStore.stores.map(store => {
-      val aclChangeListener = new ZkNodeChangeNotificationListener(
-        zkClient, store.aclChangePath, AclChangeNotificationSequenceZNode.SequenceNumberPrefix, new AclChangedNotificationHandler(store))
-
-      aclChangeListener.init()
-      aclChangeListener
-    })
-  }
-
   private def logAuditMessage(principal: KafkaPrincipal, authorized: Boolean, operation: Operation, resource: Resource, host: String) {
     def logMessage: String = {
       val authResult = if (authorized) "Allowed" else "Denied"
@@ -343,16 +344,19 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
   }
 
   private def updateAclChangedFlag(resource: Resource) {
-    zkClient.createAclChangeNotification(resource)
+    if (legacyChangeEvent)
+      zkClient.createLegacyAclChangeNotification(resource)
+    else
+      zkClient.createAclChangeNotification(resource)
   }
 
   private def backoffTime = {
     retryBackoffMs + Random.nextInt(retryBackoffJitterMs)
   }
 
-  class AclChangedNotificationHandler(store: ZkAclStore) extends NotificationHandler {
+  object AclChangedNotificationHandler extends NotificationHandler {
     override def processNotification(notificationMessage: Array[Byte]) {
-      val resource: Resource = store.decode(notificationMessage)
+      val resource: Resource = AclChangeNotificationSequenceZNode.decode(notificationMessage)
 
       inWriteLock(lock) {
         val versionedAcls = getAclsFromZk(resource)

core/src/main/scala/kafka/zk/KafkaZkClient.scala

@@ -944,9 +944,10 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
    * Creates the required zk nodes for Acl storage
    */
   def createAclPaths(): Unit = {
+    createRecursive(AclChangeNotificationZNode.path, throwIfPathExists = false)
+
     ZkAclStore.stores.foreach(store => {
       createRecursive(store.aclPath, throwIfPathExists = false)
-      createRecursive(store.aclChangePath, throwIfPathExists = false)
       ResourceType.values.foreach(resourceType => createRecursive(store.path(resourceType), throwIfPathExists = false))
     })
   }
@@ -1005,12 +1006,27 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
   }
 
   /**
-   * Creates Acl change notification message
-   * @param resource resource name
+    * Creates colon separated Acl change notification message.
+    *
+    * Kafka 2.0 saw the format of the ACL change event change from a 'colon separated' string, to a JSON message.
+    * This method will create a message in the old format, which is still needed while a cluster has
+    * 'inter.broker.protocol.version' < 2.0.
+    *
+    * @param resource resource pattern that has changed
+    */
+  def createLegacyAclChangeNotification(resource: Resource): Unit = {
+    val path = AclChangeNotificationSequenceZNode.createPath
+    val createRequest = CreateRequest(path, AclChangeNotificationSequenceZNode.encodeLegacy(resource), acls(path), CreateMode.PERSISTENT_SEQUENTIAL)
+    val createResponse = retryRequestUntilConnected(createRequest)
+    createResponse.maybeThrow
+  }
+
+  /**
+   * Creates JSON Acl change notification message.
+   * @param resource resource pattern that has changed
    */
   def createAclChangeNotification(resource: Resource): Unit = {
-    val store = ZkAclStore(resource.nameType)
-    val path = store.changeSequenceZNode.createPath
+    val path = AclChangeNotificationSequenceZNode.createPath
     val createRequest = CreateRequest(path, AclChangeNotificationSequenceZNode.encode(resource), acls(path), CreateMode.PERSISTENT_SEQUENTIAL)
     val createResponse = retryRequestUntilConnected(createRequest)
     createResponse.maybeThrow
@@ -1034,24 +1050,21 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
    * @throws KeeperException if there is an error while deleting Acl change notifications
    */
   def deleteAclChangeNotifications(): Unit = {
-    ZkAclStore.stores.foreach(store => {
-      val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(store.aclChangePath))
-      if (getChildrenResponse.resultCode == Code.OK) {
-        deleteAclChangeNotifications(store, getChildrenResponse.children)
-      } else if (getChildrenResponse.resultCode != Code.NONODE) {
-        getChildrenResponse.maybeThrow
-      }
-    })
+    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(AclChangeNotificationZNode.path))
+    if (getChildrenResponse.resultCode == Code.OK) {
+      deleteAclChangeNotifications(getChildrenResponse.children)
+    } else if (getChildrenResponse.resultCode != Code.NONODE) {
+      getChildrenResponse.maybeThrow
+    }
   }
 
   /**
    * Deletes the Acl change notifications associated with the given sequence nodes
    * @param sequenceNodes
    */
-  private def deleteAclChangeNotifications(store: ZkAclStore, sequenceNodes: Seq[String]): Unit = {
-    val aclChangeNotificationSequenceZNode = store.changeSequenceZNode
+  private def deleteAclChangeNotifications(sequenceNodes: Seq[String]): Unit = {
     val deleteRequests = sequenceNodes.map { sequenceNode =>
-      DeleteRequest(aclChangeNotificationSequenceZNode.deletePath(sequenceNode), ZkVersion.NoVersion)
+      DeleteRequest(AclChangeNotificationSequenceZNode.deletePath(sequenceNode), ZkVersion.NoVersion)
     }
 
     val deleteResponses = retryRequestsUntilConnected(deleteRequests)

core/src/main/scala/kafka/zk/ZkData.scala

@@ -30,6 +30,7 @@ import kafka.security.auth.{Acl, Resource, ResourceType}
 import kafka.server.{ConfigType, DelegationTokenManager}
 import kafka.utils.Json
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.UnsupportedVersionException
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.resource.ResourceNameType
 import org.apache.kafka.common.security.auth.SecurityProtocol
@@ -42,6 +43,7 @@ import scala.beans.BeanProperty
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.{Seq, breakOut}
+import scala.util.{Failure, Success, Try}
 
 // This file contains objects for encoding/decoding data stored in ZooKeeper nodes (znodes).
 
@@ -461,22 +463,12 @@ case class ZkAclStore(nameType: ResourceNameType) {
   val aclPath: String = nameType match {
     case ResourceNameType.LITERAL => "/kafka-acl"
     case ResourceNameType.PREFIXED => "/kafka-prefixed-acl"
-    case _ => throw new IllegalArgumentException("Unknown name type:" + nameType)
-  }
-
-  val aclChangePath: String = nameType match {
-    case ResourceNameType.LITERAL => "/kafka-acl-changes"
-    case ResourceNameType.PREFIXED => "/kafka-prefixed-acl-changes"
-    case _ => throw new IllegalArgumentException("Unknown name type:" + nameType)
+    case _ => throw new IllegalArgumentException("Invalid name type:" + nameType)
   }
 
   def path(resourceType: ResourceType) = s"$aclPath/$resourceType"
 
   def path(resourceType: ResourceType, resourceName: String): String = s"$aclPath/$resourceType/$resourceName"
-
-  def changeSequenceZNode: AclChangeNotificationSequenceZNode = AclChangeNotificationSequenceZNode(this)
-
-  def decode(notificationMessage: Array[Byte]): Resource = AclChangeNotificationSequenceZNode.decode(nameType, notificationMessage)
 }
 
 object ZkAclStore {
@@ -485,7 +477,7 @@ object ZkAclStore {
     .map(nameType => ZkAclStore(nameType))
 
   val securePaths: Seq[String] = stores
-    .flatMap(store => List(store.aclPath, store.aclChangePath))
+    .flatMap(store => List(store.aclPath)) ++ Seq(AclChangeNotificationZNode.path)
 }
 
 object ResourceZNode {
@@ -495,26 +487,49 @@ object ResourceZNode {
   def decode(bytes: Array[Byte], stat: Stat): VersionedAcls = VersionedAcls(Acl.fromBytes(bytes), stat.getVersion)
 }
 
+object AclChangeNotificationZNode {
+  def path = "/kafka-acl-changes"
+}
+
+case class AclChangeEvent(@BeanProperty @JsonProperty("version") version: Int,
+                          @BeanProperty @JsonProperty("resourceType") resourceType: String,
+                          @BeanProperty @JsonProperty("name") name: String,
+                          @BeanProperty @JsonProperty("resourceNameType") resourceNameType: String) {
+  if (version > AclChangeEvent.currentVersion)
+    throw new UnsupportedVersionException(s"Acl change event received for unsupported version: $version")
+
+  def toResource : Try[Resource] = {
+    for {
+      resType <- Try(ResourceType.fromString(resourceType))
+      nameType <- Try(ResourceNameType.valueOf(resourceNameType))
+      resource = Resource(resType, name, nameType)
+    } yield resource
+  }
+}
+
+object AclChangeEvent {
+  val currentVersion: Int = 1
+}
+
 object AclChangeNotificationSequenceZNode {
-  val Separator = ":"
   def SequenceNumberPrefix = "acl_changes_"
 
-  def encode(resource: Resource): Array[Byte] = {
-    (resource.resourceType.name + Separator + resource.name).getBytes(UTF_8)
-  }
+  def createPath = s"${AclChangeNotificationZNode.path}/$SequenceNumberPrefix"
+  def deletePath(sequenceNode: String) = s"${AclChangeNotificationZNode.path}/$sequenceNode"
 
-  def decode(nameType: ResourceNameType, bytes: Array[Byte]): Resource = {
-    val str = new String(bytes, UTF_8)
-    str.split(Separator, 2) match {
-      case Array(resourceType, name, _*) => Resource(ResourceType.fromString(resourceType), name, nameType)
-      case _ => throw new IllegalArgumentException("expected a string in format ResourceType:ResourceName but got " + str)
-    }
+  def encodeLegacy(resource: Resource): Array[Byte] = {
+    if (resource.nameType != ResourceNameType.LITERAL)
+      throw new IllegalArgumentException("Only literal resource patterns can be encoded")
+
+    val legacyName = resource.resourceType + Resource.Separator + resource.name
+    legacyName.getBytes(UTF_8)
   }
-}
 
-case class AclChangeNotificationSequenceZNode(store: ZkAclStore) {
-  def createPath = s"${store.aclChangePath}/${AclChangeNotificationSequenceZNode.SequenceNumberPrefix}"
-  def deletePath(sequenceNode: String) = s"${store.aclChangePath}/$sequenceNode"
+  def encode(resource: Resource): Array[Byte] =
+    resource.toString.getBytes(UTF_8)
+
+  def decode(bytes: Array[Byte]): Resource =
+    Resource.fromString(new String(bytes, UTF_8))
 }
 
 object ClusterZNode {

core/src/test/scala/kafka/security/auth/ResourceTest.scala

@@ -24,10 +24,15 @@ import org.junit.Assert._
 
 class ResourceTest {
   @Test(expected = classOf[KafkaException])
-  def shouldThrowTwoPartStringWithUnknownResourceType(): Unit = {
+  def shouldThrowOnTwoPartStringWithUnknownResourceType(): Unit = {
     Resource.fromString("Unknown:fred")
   }
 
+  @Test(expected = classOf[KafkaException])
+  def shouldThrowOnBadResourceTypeSeparator(): Unit = {
+    Resource.fromString("Topic-fred")
+  }
+
   @Test
   def shouldParseOldTwoPartString(): Unit = {
     assertEquals(Resource(Group, "fred", LITERAL), Resource.fromString("Group:fred"))
@@ -41,14 +46,14 @@ class ResourceTest {
 
   @Test
   def shouldParseThreePartString(): Unit = {
-    assertEquals(Resource(Group, "fred", PREFIXED), Resource.fromString("PREFIXED:Group:fred"))
-    assertEquals(Resource(Topic, "t", LITERAL), Resource.fromString("LITERAL:Topic:t"))
+    assertEquals(Resource(Group, "fred", PREFIXED), Resource.fromString("Group:PREFIXED:fred"))
+    assertEquals(Resource(Topic, "t", LITERAL), Resource.fromString("Topic:LITERAL:t"))
   }
 
   @Test
   def shouldParseThreePartWithEmbeddedSeparators(): Unit = {
-    assertEquals(Resource(Group, ":This:is:a:weird:group:name:", PREFIXED), Resource.fromString("PREFIXED:Group::This:is:a:weird:group:name:"))
-    assertEquals(Resource(Group, ":This:is:a:weird:group:name:", LITERAL), Resource.fromString("LITERAL:Group::This:is:a:weird:group:name:"))
+    assertEquals(Resource(Group, ":This:is:a:weird:group:name:", PREFIXED), Resource.fromString("Group:PREFIXED::This:is:a:weird:group:name:"))
+    assertEquals(Resource(Group, ":This:is:a:weird:group:name:", LITERAL), Resource.fromString("Group:LITERAL::This:is:a:weird:group:name:"))
   }
 
   @Test