public Mono<Void> syncAclWithAclCsv(KafkaCluster cluster, String csv) {
    return adminClientService.get(cluster)
        .flatMap(ac -> ac.listAcls(ResourcePatternFilter.ANY).flatMap(existingAclList -> {
          var existingSet = Set.copyOf(existingAclList);
          var newAcls = Set.copyOf(AclCsv.parseCsv(csv));
          var toDelete = Sets.difference(existingSet, newAcls);
          var toAdd = Sets.difference(newAcls, existingSet);
          logAclSyncPlan(cluster, toAdd, toDelete);
          if (toAdd.isEmpty() && toDelete.isEmpty()) {
            return Mono.empty();
          }
          log.info("Starting new ACLs creation");
          return ac.createAcls(toAdd)
              .doOnSuccess(v -> {
                log.info("{} new ACLs created", toAdd.size());
                log.info("Starting ACLs deletion");
              })
              .then(ac.deleteAcls(toDelete)
                  .doOnSuccess(v -> log.info("{} ACLs deleted", toDelete.size())));
        }));
  }
@Test
  void testSyncAclWithAclCsv() {
    var existingBinding1 = new AclBinding(
        new ResourcePattern(ResourceType.TOPIC, "*", PatternType.LITERAL),
        new AccessControlEntry("User:test1", "*", AclOperation.READ, AclPermissionType.ALLOW));

    var existingBinding2 = new AclBinding(
        new ResourcePattern(ResourceType.GROUP, "group1", PatternType.PREFIXED),
        new AccessControlEntry("User:test2", "localhost", AclOperation.DESCRIBE, AclPermissionType.DENY));

    var newBindingToBeAdded = new AclBinding(
        new ResourcePattern(ResourceType.GROUP, "groupNew", PatternType.PREFIXED),
        new AccessControlEntry("User:test3", "localhost", AclOperation.DESCRIBE, AclPermissionType.DENY));

    when(adminClientMock.listAcls(ResourcePatternFilter.ANY))
        .thenReturn(Mono.just(List.of(existingBinding1, existingBinding2)));

    ArgumentCaptor<Collection<AclBinding>> createdCaptor = ArgumentCaptor.forClass(Collection.class);
    when(adminClientMock.createAcls(createdCaptor.capture()))
        .thenReturn(Mono.empty());

    ArgumentCaptor<Collection<AclBinding>> deletedCaptor = ArgumentCaptor.forClass(Collection.class);
    when(adminClientMock.deleteAcls(deletedCaptor.capture()))
        .thenReturn(Mono.empty());

    aclsService.syncAclWithAclCsv(
        CLUSTER,
        "Principal,ResourceType, PatternType, ResourceName,Operation,PermissionType,Host\n"
            + "User:test1,TOPIC,LITERAL,*,READ,ALLOW,*\n"
            + "User:test3,GROUP,PREFIXED,groupNew,DESCRIBE,DENY,localhost"
    ).block();

    Collection<AclBinding> createdBindings = createdCaptor.getValue();
    assertThat(createdBindings)
        .hasSize(1)
        .contains(newBindingToBeAdded);

    Collection<AclBinding> deletedBindings = deletedCaptor.getValue();
    assertThat(deletedBindings)
        .hasSize(1)
        .contains(existingBinding2);
  }