KAFKA-13889: Fix AclsDelta to handle ACCESS_CONTROL_ENTRY_RECORD quickly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL #12160

andymg3 · 2022-05-13T17:18:51Z

JIRA

https://issues.apache.org/jira/browse/KAFKA-13889

Description

Fixes AclsDelta to handle ACCESS_CONTROL_ENTRY_RECORD quickly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL
As explained in the JIRA, in https://github.com/apache/kafka/blob/trunk/metadata/src/main/java/org/apache/kafka/image/AclsDelta.java#L64 we store the pending deletion in the changes Map. This could override a creation that might have just happened. This is an issue because in BrokerMetadataPublisher this results in us making a removeAcl call which finally results in https://github.com/apache/kafka/blob/trunk/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java#L203 being executed and this code throws an exception if the ACL isnt in the Map yet. If the ACCESS_CONTROL_ENTRY_RECORD event never got processed by BrokerMetadataPublisher then the ACL wont be in the Map yet.
So the fix here is to remove the entry from the changes Map if the ACL doesnt exist in the image yet.

Testing

Added unit tests for new behavior

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

jsancio

Thanks for the improvements @andymg3

jsancio · 2022-05-18T17:26:06Z

metadata/src/main/java/org/apache/kafka/image/AclsDelta.java

+        if (image.acls().containsKey(record.id())) {
+            changes.put(record.id(), Optional.empty());
+        } else if (changes.containsKey(record.id())) {
+            changes.remove(record.id());


Btw, the Jira mentions that we don't have "remove acl" to be idempotent. "Remove acl" is idempotent if they are process in the same batch/delta.

Do you mind clarifying? With the exception thrown below I think ACL removal is still not idempotent. If an ACL is added, then removed, then removed again, then the exception below will be thrown as it would have been removed from the Map.

You are correct. I missed the else if.

jsancio · 2022-05-18T17:26:42Z

metadata/src/main/java/org/apache/kafka/image/AclsDelta.java

+        } else if (changes.containsKey(record.id())) {
+            changes.remove(record.id());
+        } else {
+            throw new RuntimeException("Failed to find existing ACL with ID " + record.id() + " in either image or changes");


Can we throw an IllegalStateException?

I agree that exception is more appropriate. I went with RuntimeException because thats what we throw in AclControlManager (https://github.com/apache/kafka/blob/trunk/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java#L202) when we come across a removal unexpectedly. My thought process was to be consistent. Having said that, they are two different components so I'm fine updating it here to IllegalStateException. Do you think we should throw the same exception in AclControlManager?

jsancio · 2022-05-18T17:27:22Z

metadata/src/main/java/org/apache/kafka/image/AclsDelta.java

@@ -61,7 +61,13 @@ public void replay(AccessControlEntryRecord record) {
    }

    public void replay(RemoveAccessControlEntryRecord record) {


Can we document this method? Seems subtle enough to warrant documentation.

Will add a java doc to this method.

jsancio · 2022-05-18T17:29:12Z

metadata/src/main/java/org/apache/kafka/image/AclsDelta.java

+            changes.remove(record.id());
+        } else {
+            throw new RuntimeException("Failed to find existing ACL with ID " + record.id() + " in either image or changes");
+        }


This comment is for:

public Map<Uuid, Optional<StandardAcl>> changes() { return changes; }

Can we document this method? I am particularly interested on the return type. I get the impression that a value of Optional.empty should interpreted as a delete.

That's correct and that's the existing behavior. I'll add a comment clarifying that.

jsancio · 2022-05-18T17:33:29Z

metadata/src/test/java/org/apache/kafka/image/AclsDeltaTest.java

+import java.util.Optional;
+
+@Timeout(40)
+public class AclsDeltaTest {


I am okay moving these test to AclImageTest. The two types are closely related. We already have "delta" tests in the "image" suite.

Hmm yeah that's true. I guess I don't see a big downside of having a separate test class as obviously it is different to some degree. So will keep the separate test class unless you have concerns.

jsancio · 2022-05-18T17:35:18Z

metadata/src/test/java/org/apache/kafka/image/AclsDeltaTest.java

+
+        delta.replay(inputAclRecord);
+        assertTrue(delta.changes().containsKey(aclId));
+        assertTrue(delta.changes().get(aclId).isPresent());


I think you can remove this since you are implicitly testing this in the line below.

True but delta.changes().get(aclId) could return null in which case we'd throw a NPE which would be a bit more challenging to debug from the test logs in my view so will leave as is unless you have concerns.

Agree but now that you have assertEquals(Optional.of(testStandardAcl()), delta.changes().get(aclId)); there shouldn't be a NPE, right?

Good point. I think I can get rid of those two above assertions now. I just tested this with an invalid input to the Map and saw:

AclsDeltaTest > testRemovesDeleteIfNotInImage() FAILED org.opentest4j.AssertionFailedError: expected: <Optional[StandardAcl(resourceType=ANY, resourceName=foo, patternType=ANY, principal=User:user, host=host, operation=ANY, permissionType=ANY)]> but was: <null>

So we should be safe from seeing a NPE. Will update.

jsancio · 2022-05-18T17:37:12Z

metadata/src/test/java/org/apache/kafka/image/AclsDeltaTest.java

+        delta.replay(inputAclRecord);
+        assertTrue(delta.changes().containsKey(aclId));
+        assertTrue(delta.changes().get(aclId).isPresent());
+        assertEquals(testStandardAcl(), delta.changes().get(aclId).get());


You can remove the get() and compare it against Optional.of(testStandardAcl()).

Agree that's cleaner and it prevents NPEs if delta.changes().get(aclId) returns null

jsancio · 2022-05-18T17:38:45Z

metadata/src/test/java/org/apache/kafka/image/AclsDeltaTest.java

+    }
+
+    @Test
+    public void testThrowsExceptionOnInvalidState() {


Can we also test with an AclsImage that has ACLs but for the one being removed?

jsancio

Thanks for the update. Just one minor comment.

jsancio · 2022-05-18T21:14:18Z

metadata/src/main/java/org/apache/kafka/image/AclsDelta.java

+        if (image.acls().containsKey(record.id())) {
+            changes.put(record.id(), Optional.empty());
+        } else if (changes.containsKey(record.id())) {
+            changes.remove(record.id());


You are correct. I missed the else if.

jsancio · 2022-05-18T21:16:33Z

metadata/src/test/java/org/apache/kafka/image/AclsDeltaTest.java

+
+        delta.replay(inputAclRecord);
+        assertTrue(delta.changes().containsKey(aclId));
+        assertTrue(delta.changes().get(aclId).isPresent());


Agree but now that you have assertEquals(Optional.of(testStandardAcl()), delta.changes().get(aclId)); there shouldn't be a NPE, right?

…kly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL

jsancio

LGTM. The test failures seem unrelated:

Build / JDK 8 and Scala 2.12 / testSeekAndCommitWithBrokerFailures() – kafka.api.ConsumerBounceTest3s
Build / ARM / testListenerConnectionRateLimitWhenActualRateAboveLimit() – kafka.network.ConnectionQuotasTest

andymg3 force-pushed the KAFKA-13889 branch 3 times, most recently from 57a6b08 to 3ef6626 Compare May 18, 2022 13:59

jsancio self-requested a review May 18, 2022 16:51

jsancio reviewed May 18, 2022

View reviewed changes

andymg3 force-pushed the KAFKA-13889 branch from 3ef6626 to 854979c Compare May 18, 2022 20:09

jsancio reviewed May 18, 2022

View reviewed changes

andymg3 added 3 commits May 19, 2022 10:44

KAFKA-13889: Fix AclsDelta to handle ACCESS_CONTROL_ENTRY_RECORD quic…

d782c62

…kly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL

Address PR feedback

954de1d

Remove unnecessary assertions

86e62aa

andymg3 force-pushed the KAFKA-13889 branch from 342990a to 86e62aa Compare May 19, 2022 14:46

jsancio approved these changes May 19, 2022

View reviewed changes

jsancio merged commit a8e3a25 into apache:trunk May 19, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-13889: Fix AclsDelta to handle ACCESS_CONTROL_ENTRY_RECORD quickly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL #12160

KAFKA-13889: Fix AclsDelta to handle ACCESS_CONTROL_ENTRY_RECORD quickly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL #12160

andymg3 commented May 13, 2022

jsancio left a comment

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio May 18, 2022

andymg3 May 18, 2022

jsancio left a comment

jsancio May 18, 2022

jsancio May 18, 2022

jsancio left a comment

		@@ -61,7 +61,13 @@ public void replay(AccessControlEntryRecord record) {
		}

		public void replay(RemoveAccessControlEntryRecord record) {

KAFKA-13889: Fix AclsDelta to handle ACCESS_CONTROL_ENTRY_RECORD quickly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL #12160

KAFKA-13889: Fix AclsDelta to handle ACCESS_CONTROL_ENTRY_RECORD quickly followed by REMOVE_ACCESS_CONTROL_ENTRY_RECORD for same ACL #12160

Conversation

andymg3 commented May 13, 2022

JIRA

Description

Testing

Committer Checklist (excluded from commit message)

jsancio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio left a comment

Choose a reason for hiding this comment