KAFKA-15983 Kafka-acls should return authorization already done if repeating work is issued (apache#20482)

# Description
`kafka-acls.sh` doesn't print message about duplicate authorization.

# Changes 
Now the cli searches for existing AclBinding, prints duplicate bindings,
and removes them from the adding list.

Reviewers: Chia-Ping Tsai <[email protected]>

Author: jack2012aa

tools/src/main/java/org/apache/kafka/tools/AclCommand.java

@@ -106,9 +106,24 @@ private static void addAcls(Admin admin, AclCommandOptions opts) throws Executio
         for (Map.Entry<ResourcePattern, Set<AccessControlEntry>> entry : resourceToAcl.entrySet()) {
             ResourcePattern resource = entry.getKey();
             Set<AccessControlEntry> acls = entry.getValue();
-            System.out.println("Adding ACLs for resource `" + resource + "`: " + NL + " " + acls.stream().map(a -> "\t" + a).collect(Collectors.joining(NL)) + NL);
-            Collection<AclBinding> aclBindings = acls.stream().map(acl -> new AclBinding(resource, acl)).collect(Collectors.toList());
-            admin.createAcls(aclBindings).all().get();
+
+            AclBindingFilter filter = new AclBindingFilter(resource.toFilter(), AccessControlEntryFilter.ANY);
+            Set<AclBinding> existingBindingsSet = Set.copyOf(admin.describeAcls(filter).values().get());
+
+            List<AclBinding> aclBindings = new ArrayList<>();
+            for (AccessControlEntry acl : acls) {
+                AclBinding binding = new AclBinding(resource, acl);
+                if (existingBindingsSet.contains(binding)) {
+                    System.out.println("Acl " + binding + " already exists.");
+                } else {
+                    aclBindings.add(binding);
+                }
+            }
+
+            if (!aclBindings.isEmpty()) {
+                System.out.println("Adding ACLs for resource `" + resource + "`: " + NL + " " + aclBindings.stream().map(AclBinding::entry).map(a -> "\t" + a).collect(Collectors.joining(NL)) + NL);
+                admin.createAcls(aclBindings).all().get();
+            }
         }
     }
 

tools/src/test/java/org/apache/kafka/tools/AclCommandTest.java

@@ -17,6 +17,7 @@
 package org.apache.kafka.tools;
 
 import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AclBinding;
 import org.apache.kafka.common.acl.AclBindingFilter;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.acl.AclPermissionType;
@@ -79,6 +80,7 @@
 import static org.apache.kafka.server.config.ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -275,6 +277,24 @@ public void testPatternTypesWithAdminAPIAndBootstrapController(ClusterInstance c
         testPatternTypes(adminArgsWithBootstrapController(cluster.bootstrapControllers(), Optional.empty()));
     }
 
+    @ClusterTest
+    public void testDuplicateAdd(ClusterInstance cluster) {
+        final String topicName = "test-topic";
+        final String principal = "User:Alice";
+        ResourcePattern resource = new ResourcePattern(ResourceType.TOPIC, topicName, PatternType.LITERAL);
+        AccessControlEntry ace = new AccessControlEntry(principal, WILDCARD_HOST, READ, ALLOW);
+        AclBinding binding = new AclBinding(resource, ace);
+        List<String> cmdArgs = adminArgs(cluster.bootstrapServers(), Optional.empty());
+        List<String> initialAddArgs = new ArrayList<>(cmdArgs);
+        initialAddArgs.addAll(List.of(ADD, TOPIC, topicName, "--allow-principal", principal, OPERATION, "Read"));
+
+        callMain(initialAddArgs);
+        String out = callMain(initialAddArgs).getKey();
+
+        assertTrue(out.contains("Acl " + binding + " already exists."));
+        assertFalse(out.contains("Adding ACLs for resource"));
+    }
+
     @Test
     public void testUseBootstrapServerOptWithBootstrapControllerOpt() {
         assertInitializeInvalidOptionsExitCodeAndMsg(