MINOR: Restore original behavior of GroupAuthorizerIntegrationTest

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/security/GroupAuthorizerIntegrationTest.java

@@ -141,7 +141,9 @@ private AccessControlEntry createAcl(AclOperation aclOperation, AclPermissionTyp
     }
 
     private void addAndVerifyAcls(Set<AccessControlEntry> acls, ResourcePattern resource, ClusterInstance clusterInstance) throws InterruptedException {
-        List<AclBinding> aclBindings = acls.stream().map(acl -> new AclBinding(resource, acl)).toList();
+        List<AclBinding> aclBindings = acls.stream()
+                .map(acl -> new AclBinding(resource, acl))
+                .toList();
         Authorizer authorizer = getAuthorizer(clusterInstance);
         authorizer.createAcls(ANONYMOUS_CONTEXT, aclBindings)
                 .forEach(future -> {
@@ -155,6 +157,29 @@ private void addAndVerifyAcls(Set<AccessControlEntry> acls, ResourcePattern reso
         clusterInstance.waitAcls(aclBindingFilter, acls);
     }
 
+    private void removeAndVerifyAcls(Set<AccessControlEntry> deleteAcls, ResourcePattern resource, ClusterInstance clusterInstance) throws InterruptedException {
+        List<AclBindingFilter> aclBindingFilters = deleteAcls.stream()
+                .map(acl -> new AclBindingFilter(resource.toFilter(), acl.toFilter()))
+                .toList();
+        Authorizer authorizer = getAuthorizer(clusterInstance);
+        authorizer.deleteAcls(ANONYMOUS_CONTEXT, aclBindingFilters)
+                .forEach(future -> {
+                    try {
+                        future.toCompletableFuture().get();
+                    } catch (InterruptedException | ExecutionException e) {
+                        throw new RuntimeException("Failed to delete ACLs", e);
+                    }
+                });
+
+        AclBindingFilter aclBindingFilter = new AclBindingFilter(resource.toFilter(), AccessControlEntryFilter.ANY);
+        TestUtils.waitForCondition(() -> {
+            Set<AccessControlEntry> remainingAclEntries = new HashSet<>();
+            authorizer.acls(aclBindingFilter).forEach(aclBinding -> remainingAclEntries.add(aclBinding.entry()));
+            return deleteAcls.stream().noneMatch(remainingAclEntries::contains);
+        }, "Failed to verify ACLs deletion");
+    }
+
+
     static final AuthorizableRequestContext ANONYMOUS_CONTEXT = new AuthorizableRequestContext() {
         @Override
         public String listenerName() {
@@ -253,17 +278,27 @@ public void testUnauthorizedProduceAndConsume(ClusterInstance clusterInstance, G
         }
     }
 
+    @ClusterTest
+    public void testClassicConsumeUnsubscribeWithGroupPermission(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
+        testConsumeUnsubscribeWithOrWithoutGroupPermission(clusterInstance, GroupProtocol.CLASSIC, true);
+    }
+
+    @ClusterTest
+    public void testAsyncConsumeUnsubscribeWithGroupPermission(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
+        testConsumeUnsubscribeWithOrWithoutGroupPermission(clusterInstance, GroupProtocol.CONSUMER, true);
+    }
+
     @ClusterTest
     public void testClassicConsumeUnsubscribeWithoutGroupPermission(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
-        testConsumeUnsubscribeWithGroupPermission(clusterInstance, GroupProtocol.CLASSIC);
+        testConsumeUnsubscribeWithOrWithoutGroupPermission(clusterInstance, GroupProtocol.CLASSIC, false);
     }
 
     @ClusterTest
     public void testAsyncConsumeUnsubscribeWithoutGroupPermission(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
-        testConsumeUnsubscribeWithGroupPermission(clusterInstance, GroupProtocol.CONSUMER);
+        testConsumeUnsubscribeWithOrWithoutGroupPermission(clusterInstance, GroupProtocol.CONSUMER, false);
     }
 
-    private void testConsumeUnsubscribeWithGroupPermission(ClusterInstance clusterInstance, GroupProtocol groupProtocol) throws InterruptedException, ExecutionException {
+    private void testConsumeUnsubscribeWithOrWithoutGroupPermission(ClusterInstance clusterInstance, GroupProtocol groupProtocol, boolean withGroupPermission) throws InterruptedException, ExecutionException {
         setup(clusterInstance);
         String topic = "topic";
         String group = "group";
@@ -297,6 +332,13 @@ private void testConsumeUnsubscribeWithGroupPermission(ClusterInstance clusterIn
                 ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofSeconds(15));
                 return records.count() == 1;
             }, "consumer failed to receive message");
+            if (!withGroupPermission) {
+                removeAndVerifyAcls(
+                        Set.of(createAcl(AclOperation.READ, AclPermissionType.ALLOW, CLIENT_PRINCIPAL)),
+                        new ResourcePattern(ResourceType.GROUP, group, PatternType.LITERAL),
+                        clusterInstance
+                );
+            }
             assertDoesNotThrow(consumer::unsubscribe);
         }
     }