Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
junrao merged 12 commits into from
Jun 12, 2018
Merged

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
16f5597
KAFKA-7007: All ACL changes should use single /kafka-acl-changes path
big-andy-coates Jun 7, 2018
259df49
Merge branch 'trunk' into single_acl_change_path
big-andy-coates Jun 9, 2018
6013925
Removed unused from ZkData
big-andy-coates Jun 9, 2018
b29bd45
Merge branch 'trunk' into single_acl_change_path
big-andy-coates Jun 9, 2018
237db81
Post merge fix up
big-andy-coates Jun 9, 2018
9185513
Jun's requested changes
big-andy-coates Jun 9, 2018
5769d56
Revert back from JSON ACL change event to String based, with either 2…
big-andy-coates Jun 10, 2018
7046b0c
Jun's change requests
big-andy-coates Jun 11, 2018
b919bd1
Switch to a hybrid:
big-andy-coates Jun 12, 2018
3280303
Remove todo
big-andy-coates Jun 12, 2018
ae50f4e
Split ZkAclChangeStore from ZkAclStore, so that we can have two ZkAcl…
big-andy-coates Jun 12, 2018
23bb972
Jun's requested changes.
big-andy-coates Jun 12, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 7 additions & 16 deletions core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ 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.resource.{ResourceFilter, ResourceNameType => JResourceNameType}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{SecurityUtils, Time}
Expand Down Expand Up @@ -55,7 +55,7 @@ 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 = _

@volatile
private var aclCache = new scala.collection.immutable.TreeMap[Resource, VersionedAcls]()(ResourceOrdering)
Expand Down Expand Up @@ -98,7 +98,8 @@ class SimpleAclAuthorizer extends Authorizer with Logging {

loadCache()

startZkChangeListeners()
aclChangeListener = new ZkNodeChangeNotificationListener(zkClient, AclChangeNotificationZNode.path, AclChangeNotificationSequenceZNode.SequenceNumberPrefix, AclChangedNotificationHandler)
aclChangeListener.init()
}

override def authorize(session: Session, operation: Operation, resource: Resource): Boolean = {
Expand Down Expand Up @@ -230,7 +231,7 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
}

def close() {
aclChangeListeners.foreach(listener => listener.close())
if (aclChangeListener != null) aclChangeListener.close()
if (zkClient != null) zkClient.close()
}

Expand All @@ -250,16 +251,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"
Expand Down Expand Up @@ -349,9 +340,9 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
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)
Expand Down
25 changes: 11 additions & 14 deletions core/src/main/scala/kafka/zk/KafkaZkClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -943,9 +943,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))
})
}
Expand Down Expand Up @@ -1008,8 +1009,7 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
* @param resource resource name
*/
def createAclChangeNotification(resource: Resource): Unit = {
val store = ZkAclStore(resource.resourceNameType)
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
Expand All @@ -1033,24 +1033,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)
Expand Down
82 changes: 56 additions & 26 deletions core/src/main/scala/kafka/zk/ZkData.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import kafka.security.auth.{Acl, Literal, Prefixed, Resource, ResourceNameType,
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.security.auth.SecurityProtocol
import org.apache.kafka.common.security.token.delegation.{DelegationToken, TokenInformation}
Expand All @@ -41,6 +42,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}

@junrao junrao Jun 11, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

unused imports Failure, Success

@big-andy-coates big-andy-coates Jun 11, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.


// This file contains objects for encoding/decoding data stored in ZooKeeper nodes (znodes).

Expand Down Expand Up @@ -460,30 +462,30 @@ case class ZkAclStore(nameType: ResourceNameType) {
val aclPath: String = nameType match {
case Literal => "/kafka-acl"
case Prefixed => "/kafka-prefixed-acl"
case _ => throw new IllegalArgumentException("Unknown name type:" + nameType)
}

val aclChangePath: String = nameType match {
case Literal => "/kafka-acl-changes"
case 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 {
val stores: Seq[ZkAclStore] = ResourceNameType.values
.map(nameType => ZkAclStore(nameType))

val securePaths: Seq[String] = stores
.flatMap(store => List(store.aclPath, store.aclChangePath))
.flatMap(store => List(store.aclPath)) ++ Seq(AclChangeNotificationZNode.path)
}

@deprecated("There are now multiple roots for ACLs within ZK. Use ZkAclStore", "2.0")
object AclZNode {
def path = "/kafka-acl"
}

@deprecated("There are now multiple roots for ACLs within ZK. Use ZkAclStore", "2.0")
object ResourceTypeZNode {

@junrao junrao Jun 7, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AclZNode and ResourceTypeZNode are not part of the public API. We can just remove them.

@big-andy-coates big-andy-coates Jun 8, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

def path(resourceType: String) = s"${AclZNode.path}/$resourceType"
}

object ResourceZNode {
Expand All @@ -493,26 +495,54 @@ 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,

@junrao junrao Jun 11, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This class is no longer used.

@big-andy-coates big-andy-coates Jun 11, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

@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.fromString(resourceNameType))
resource = Resource(resType, name, nameType)
} yield resource
}
}

object AclChangeEvent {
val currentVersion: Int = 1

@junrao junrao Jun 11, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is no longer needed?

@big-andy-coates big-andy-coates Jun 11, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, and done.

}

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 encode(resource: Resource): Array[Byte] =
Json.encodeAsBytes(AclChangeEvent(
AclChangeEvent.currentVersion,
resource.resourceType.name,
resource.name,
resource.resourceNameType.name))

def decode(bytes: Array[Byte]): Resource = {
val changeEvent = Json.parseBytesAs[AclChangeEvent](bytes) match {
case Right(event) => event
case Left(e) => throw new IllegalArgumentException("Failed to parse ACL change event", e)

@junrao junrao Jun 7, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

During the upgrade phase, the new broker may see an acl_change node of the old format. So, we want to fall back to parse the value of the node using the old format if it's not json.

@big-andy-coates big-andy-coates Jun 9, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

}
}
}

case class AclChangeNotificationSequenceZNode(store: ZkAclStore) {
def createPath = s"${store.aclChangePath}/${AclChangeNotificationSequenceZNode.SequenceNumberPrefix}"
def deletePath(sequenceNode: String) = s"${store.aclChangePath}/$sequenceNode"
changeEvent.toResource match {
case Success(r) => r
case Failure(e) => throw new IllegalArgumentException("Failed to parse ACL change event", e)
}
}
}

object ClusterZNode {
Expand Down
6 changes: 3 additions & 3 deletions core/src/test/scala/unit/kafka/common/ZkNodeChangeNotificationListenerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ package kafka.common

import kafka.security.auth.{Group, Literal, Resource}
import kafka.utils.TestUtils
import kafka.zk.{AclChangeNotificationSequenceZNode, ZkAclStore, ZooKeeperTestHarness}
import kafka.zk.{AclChangeNotificationSequenceZNode, AclChangeNotificationZNode, ZooKeeperTestHarness}
import org.junit.{After, Test}

class ZkNodeChangeNotificationListenerTest extends ZooKeeperTestHarness {
Expand All @@ -38,7 +38,7 @@ class ZkNodeChangeNotificationListenerTest extends ZooKeeperTestHarness {
@volatile var invocationCount = 0
val notificationHandler = new NotificationHandler {
override def processNotification(notificationMessage: Array[Byte]): Unit = {
notification = AclChangeNotificationSequenceZNode.decode(Literal, notificationMessage)
notification = AclChangeNotificationSequenceZNode.decode(notificationMessage)
invocationCount += 1
}
}
Expand All @@ -48,7 +48,7 @@ class ZkNodeChangeNotificationListenerTest extends ZooKeeperTestHarness {
val notificationMessage2 = Resource(Group, "messageB", Literal)
val changeExpirationMs = 1000

notificationListener = new ZkNodeChangeNotificationListener(zkClient, ZkAclStore(Literal).aclChangePath,
notificationListener = new ZkNodeChangeNotificationListener(zkClient, AclChangeNotificationZNode.path,
AclChangeNotificationSequenceZNode.SequenceNumberPrefix, notificationHandler, changeExpirationMs)
notificationListener.init()

Expand Down
16 changes: 8 additions & 8 deletions core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,19 +426,19 @@ class KafkaZkClientTest extends ZooKeeperTestHarness {

@Test
def testAclManagementMethods() {

assertFalse(zkClient.pathExists(AclChangeNotificationZNode.path))
ZkAclStore.stores.foreach(store => {
assertFalse(zkClient.pathExists(store.aclPath))
assertFalse(zkClient.pathExists(store.aclChangePath))
ResourceType.values.foreach(resource => assertFalse(zkClient.pathExists(store.path(resource))))
})

// create acl paths
zkClient.createAclPaths

assertTrue(zkClient.pathExists(AclChangeNotificationZNode.path))

ZkAclStore.stores.foreach(store => {
assertTrue(zkClient.pathExists(store.aclPath))
assertTrue(zkClient.pathExists(store.aclChangePath))
ResourceType.values.foreach(resource => assertTrue(zkClient.pathExists(store.path(resource))))

val resource1 = new Resource(Topic, UUID.randomUUID().toString, store.nameType)
Expand Down Expand Up @@ -488,15 +488,15 @@ class KafkaZkClientTest extends ZooKeeperTestHarness {
//delete with valid expected zk version
assertTrue(zkClient.conditionalDelete(resource2, 0))


zkClient.createAclChangeNotification(Resource(Group, "resource1", store.nameType))
zkClient.createAclChangeNotification(Resource(Topic, "resource2", store.nameType))
})

assertEquals(2, zkClient.getChildren(store.aclChangePath).size)
val expectedChangeEvents = ResourceNameType.values.size * 2
assertEquals(expectedChangeEvents, zkClient.getChildren(AclChangeNotificationZNode.path).size)

zkClient.deleteAclChangeNotifications()
assertTrue(zkClient.getChildren(store.aclChangePath).isEmpty)
})
zkClient.deleteAclChangeNotifications()
assertTrue(zkClient.getChildren(AclChangeNotificationZNode.path).isEmpty)
}

@Test
Expand Down
7 changes: 4 additions & 3 deletions docs/upgrade.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ <h4><a id="upgrade_2_0_0" href="#upgrade_2_0_0">Upgrading from 0.8.x, 0.9.x, 0.1
<li>inter.broker.protocol.version=CURRENT_KAFKA_VERSION (0.11.0, 1.0, 1.1, 1.2).</li>
</ul>
</li>
<li> IMPORTANT: Do not change any ACLs during the upgrade. (The format of the ACL change event has changed and is not compatible with old brokers).</li>

@junrao junrao Jun 7, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a bit too restricting. To support adding acls during rolling upgrade, we probably only want write the acl_change path with the new json format if inter.broker.protocol is >= 2.0. Otherwise, if the new ACL doesn't use the new prefix feature, we switch to using the old format. Otherwise, we fail the addAcl request.

Since the AclCommand still writes to ZK directly. We probably want to recommend "Not use the 2.0 AclCommand tool until all brokers are upgraded to 2.0."

@big-andy-coates big-andy-coates Jun 9, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @junrao, my bad - we did discuss this. Sorry I missed this. Done now.

<li> Upgrade the brokers one at a time: shut down the broker, update the code, and restart it. </li>
<li> Once the entire cluster is upgraded, bump the protocol version by editing <code>inter.broker.protocol.version</code> and setting it to 2.0.
<li> Restart the brokers one by one for the new protocol version to take effect.</li>
Expand All @@ -61,10 +62,10 @@ <h4><a id="upgrade_2_0_0" href="#upgrade_2_0_0">Upgrading from 0.8.x, 0.9.x, 0.1
Similarly for the message format version.</li>
<li>If you are using Java8 method references in your Kafka Streams code you might need to update your code to resolve method ambiguities.
Hot-swapping the jar-file only might not work.</li>
<li>ACLs should not be added to prefixed resources,
<li>ACLs should not be added during the upgrade process.

@junrao junrao Jun 8, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, if we have upgraded the whole cluster and started using the new acl_change format in ZK, downgrading the cluster will cause the processing of acl_change sequence node to fail and the sequence node not to be deleted. This means the downgraded broker can't process ACL changes forever. So, your original idea of using a new acl_change path such as /kafka-acl-changes-v2 seems better.

@junrao junrao Jun 8, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, this has the other problem. Since the AclCommand tool still writes to ZK directly, if we switch to a new acl-change path, the old version of AclCommand won't work once the brokers are upgraded.

Given the complexity in dealing with upgrades/downgrades, maybe the current approach of using 2 separate acl-change path is not that bad. So, perhaps we can just keep the current approach, but (1) ensure that the new acl with prefix is only added if inter.broker.protocol is new; (2) In ZkNodeChangeNotificationListener.processNotifications(), catch decoding error of individual acl change value, log it and let it go. This may help future format change.

Sorry for going back and forth on this one.

@big-andy-coates big-andy-coates Jun 9, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hummm.... Good points @junrao.

I have another suggestion...

pre-2.0 behaviour uses a colon-separated string to write the change notification in the format <resource-type>:<name>, where the name can have embedded colons, e.g. a group of my:weird:group:name.

With my recent changes the toString representation of Resource is <name-type>:<resource-type>:<name>, and I've already enhanced the Resource.fromString() method to handle both old and new formats. For the solution below I'll need to swap the first two parts in the new format so that it is <resource-type>:<name-type>:<name>.

Given this new format, rather than switching to JSON we could just write out change notifications out as strings still, with either two or three parts, depending on inter.broker.protocol.version.

During a rolling upgrade change events would be written using two parts, which work with older brokers and are interpreted as 'literal' resources by the new brokers. (Attempts to add Prefixed would result in UVE).

Once upgraded brokers start to write using three parts change events.

On a downgrade older brokers would treat the 'Topic:Prefixed:Foo' as a change on a Topic called 'Prefixed:Foo', which wouldn't match anything. I'd have to check, but there's every possibility that the broker would process this without error and move on. The fact that the old broker effectively ignored the event shouldn't matter as the event must be old to be in this format.

Thoughts?

@junrao junrao Jun 9, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that seems to be a good idea. The only confusion is that if someone has a group named "Literal:weirdgroup" or "Prefixed:weirdgroup". Since that's unlikely, I think what you suggested is probably the best approach given what we have now.

@big-andy-coates big-andy-coates Jun 10, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Great. Thinking on this more, I think we should consider not adding the backwards compatibility logic, I.e. the new brokers should always write using the three part name.

If we do add this logic it will need to be in the code base for a long long time, as there could always be some user upgrading. If we don’t add it, then the edge case this affects is very small / unlikely and the impact minimal: the old broker would only gave incorrect acts until it was itself upgraded, which should be a small window.

What do you think @junrao?

<p><b>NOTE:</b> Any ACLs added to prefixed resources,
(added in <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-290%3A+Support+for+Prefixed+ACLs">KIP-290</a>),
until all brokers in the cluster have been updated.
<p><b>NOTE:</b> any prefixed ACLs added to a cluster will be ignored should the cluster be downgraded again.
will be ignored should the cluster be downgraded again.
</li>
</ol>

Expand Down