public Mono<Void> createProducerAcl(KafkaCluster cluster, CreateProducerAclDTO request) {
    return adminClientService.get(cluster)
        .flatMap(ac -> createAclsWithLogging(ac, createProducerBindings(request)))
        .then();
  }
@Test
  void createsProducerDependantAclsWhenTopicsAndTxIdSpecifiedByPrefix() {
    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.createProducerAcl(
        CLUSTER,
        new CreateProducerAclDTO()
            .principal(principal)
            .host(host)
            .topicsPrefix("topicPref")
            .transactionsIdPrefix("txIdPref")
            .idempotent(false)
    ).block();

    //Write, Describe, Create permission on topics, Write, Describe on transactionalIds
    //IDEMPOTENT_WRITE on cluster if idempotent is enabled (false)
    Collection<AclBinding> createdBindings = createdCaptor.getValue();
    assertThat(createdBindings)
        .hasSize(5)
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "topicPref", PatternType.PREFIXED),
            new AccessControlEntry(principal, host, AclOperation.WRITE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "topicPref", PatternType.PREFIXED),
            new AccessControlEntry(principal, host, AclOperation.DESCRIBE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TOPIC, "topicPref", PatternType.PREFIXED),
            new AccessControlEntry(principal, host, AclOperation.CREATE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TRANSACTIONAL_ID, "txIdPref", PatternType.PREFIXED),
            new AccessControlEntry(principal, host, AclOperation.WRITE, AclPermissionType.ALLOW)))
        .contains(new AclBinding(
            new ResourcePattern(ResourceType.TRANSACTIONAL_ID, "txIdPref", PatternType.PREFIXED),
            new AccessControlEntry(principal, host, AclOperation.DESCRIBE, AclPermissionType.ALLOW)));
  }