public Mono<Void> createStreamAppAcl(KafkaCluster cluster, CreateStreamAppAclDTO request) {
    return adminClientService.get(cluster)
        .flatMap(ac -> createAclsWithLogging(ac, createStreamAppBindings(request)))
        .then();
  }
@Test
  void createsStreamAppDependantAcls() {
    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.createStreamAppAcl(
        CLUSTER,
        new CreateStreamAppAclDTO()
            .principal(principal)
            .host(host)
            .inputTopics(List.of("t1"))
            .outputTopics(List.of("t2", "t3"))
            .applicationId("appId1")
    ).block();

    // Read on input topics, Write on output topics
    // ALL on applicationId-prefixed Groups and Topics
    Collection<AclBinding> createdBindings = createdCaptor.getValue();
    assertThat(createdBindings)
        .hasSize(5)
        .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, "t2", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.WRITE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "t3", PatternType.LITERAL),
            new AccessControlEntry(principal, host, AclOperation.WRITE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.GROUP, "appId1", PatternType.PREFIXED),
            new AccessControlEntry(principal, host, AclOperation.ALL, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "appId1", PatternType.PREFIXED),
            new AccessControlEntry(principal, host, AclOperation.ALL, AclPermissionType.ALLOW)));
  }