@Test
    public void testCreateAclDeleteAcl() {
        AclControlManager manager = new AclControlManager.Builder().build();
        MockClusterMetadataAuthorizer authorizer = new MockClusterMetadataAuthorizer();
        authorizer.loadSnapshot(manager.idToAcl());

        List<AclBinding> toCreate = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            toCreate.add(TEST_ACLS.get(i).toBinding());
        }
        toCreate.add(new AclBinding(
            new ResourcePattern(TOPIC, "*", PatternType.UNKNOWN),
            new AccessControlEntry("User:*", "*", ALTER, ALLOW)));

        ControllerResult<List<AclCreateResult>> createResult = manager.createAcls(toCreate);

        List<AclCreateResult> expectedResults = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            expectedResults.add(AclCreateResult.SUCCESS);
        }
        expectedResults.add(new AclCreateResult(
            new InvalidRequestException("Invalid patternType UNKNOWN")));

        for (int i = 0; i < expectedResults.size(); i++) {
            AclCreateResult expectedResult = expectedResults.get(i);
            if (expectedResult.exception().isPresent()) {
                assertEquals(expectedResult.exception().get().getMessage(),
                    createResult.response().get(i).exception().get().getMessage());
            } else {
                assertFalse(createResult.response().get(i).exception().isPresent());
            }
        }
        RecordTestUtils.replayAll(manager, createResult.records());
        assertFalse(manager.idToAcl().isEmpty());

        ControllerResult<List<AclDeleteResult>> deleteResult =
            manager.deleteAcls(Arrays.asList(
                new AclBindingFilter(
                    new ResourcePatternFilter(ResourceType.ANY, null, LITERAL),
                        AccessControlEntryFilter.ANY),
                new AclBindingFilter(
                    new ResourcePatternFilter(ResourceType.UNKNOWN, null, LITERAL),
                        AccessControlEntryFilter.ANY)));
        assertEquals(2, deleteResult.response().size());
        Set<AclBinding> deleted = new HashSet<>();
        for (AclDeleteResult.AclBindingDeleteResult result :
                deleteResult.response().get(0).aclBindingDeleteResults()) {
            assertEquals(Optional.empty(), result.exception());
            deleted.add(result.aclBinding());
        }
        assertEquals(new HashSet<>(Arrays.asList(
            TEST_ACLS.get(0).toBinding(),
                TEST_ACLS.get(2).toBinding())), deleted);
        assertEquals(InvalidRequestException.class,
            deleteResult.response().get(1).exception().get().getClass());
        RecordTestUtils.replayAll(manager, deleteResult.records());

        Iterator<Map.Entry<Uuid, StandardAcl>> iterator = manager.idToAcl().entrySet().iterator();
        assertEquals(TEST_ACLS.get(1).acl(), iterator.next().getValue());
        assertFalse(iterator.hasNext());
    }
Verify that validateFilter catches invalid filters.
metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java