From 9dd4cff2d79311ed763fcb750d39a6e524cabea8 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Wed, 28 May 2025 12:40:19 -0400 Subject: [PATCH] KAFKA-19347 Don't update timeline data structures in createAcls (#19840) This patch fixes a problem in AclControlManager where we are updating the timeline data structures prematurely. Reviewers: Alyssa Huang , Colin P. McCabe , Andrew Schofield , --- .../java/org/apache/kafka/controller/AclControlManager.java | 5 +++-- .../org/apache/kafka/controller/AclControlManagerTest.java | 1 + 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java index ca324ab5788..08ed04d5301 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java @@ -106,10 +106,11 @@ public class AclControlManager { continue; } StandardAcl standardAcl = StandardAcl.fromAclBinding(acl); - if (existingAcls.add(standardAcl)) { + if (!existingAcls.contains(standardAcl)) { StandardAclWithId standardAclWithId = new StandardAclWithId(newAclId(), standardAcl); - idToAcl.put(standardAclWithId.id(), standardAcl); records.add(new ApiMessageAndVersion(standardAclWithId.toRecord(), (short) 0)); + } else { + log.debug("Not creating ACL since it already exists: {}", standardAcl); } results.add(AclCreateResult.SUCCESS); } diff --git a/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java index 33a9333d475..64f193b5ef8 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java @@ -346,6 +346,7 @@ public class AclControlManagerTest { new AccessControlEntry("User:user", "10.0.0.1", AclOperation.ALL, ALLOW)); ControllerResult> createResult = manager.createAcls(List.of(aclBinding)); + RecordTestUtils.replayAll(manager, createResult.records()); Uuid id = ((AccessControlEntryRecord) createResult.records().get(0).message()).id(); assertEquals(1, createResult.records().size());