public Mono<Void> createConsumerAcl(KafkaCluster cluster, CreateConsumerAclDTO request) {
    return adminClientService.get(cluster)
        .flatMap(ac -> createAclsWithLogging(ac, createConsumerBindings(request)))
        .then();
  }
@Test
  void createsConsumerDependantAcls() {
    ArgumentCaptor<Collection<AclBinding>> createdCaptor = ArgumentCaptor.forClass(Collection.class);
    when(adminClientMock.createAcls(createdCaptor.capture()))
        .thenReturn(Mono.empty());

    var principal = UUID.randomUUID().toString();
    var host = UUID.randomUUID().toString();

    aclsService.createConsumerAcl(
        CLUSTER,
        new CreateConsumerAclDTO()
            .principal(principal)
            .host(host)
            .consumerGroups(List.of("cg1", "cg2"))
            .topics(List.of("t1", "t2"))
    ).block();

    //Read, Describe on topics, Read on consumerGroups
    Collection<AclBinding> createdBindings = createdCaptor.getValue();
    assertThat(createdBindings)
        .hasSize(6)
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "t1", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.READ, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "t1", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.DESCRIBE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "t2", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.READ, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "t2", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.DESCRIBE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.GROUP, "cg1", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.READ, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.GROUP, "cg2", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.READ, AclPermissionType.ALLOW)));
  }