public static KafkaUserModel fromCrd(KafkaUser kafkaUser,
                                         String secretPrefix,
                                         boolean aclsAdminApiSupported) {
        KafkaUserModel result = new KafkaUserModel(kafkaUser.getMetadata().getNamespace(),
                kafkaUser.getMetadata().getName(),
                Labels.fromResource(kafkaUser).withStrimziKind(kafkaUser.getKind()),
                secretPrefix);

        validateTlsUsername(kafkaUser);
        validateDesiredPassword(kafkaUser);

        result.setOwnerReference(kafkaUser);
        result.setAuthentication(kafkaUser.getSpec().getAuthentication());

        if (kafkaUser.getSpec().getAuthorization() != null && kafkaUser.getSpec().getAuthorization().getType().equals(KafkaUserAuthorizationSimple.TYPE_SIMPLE)) {
            if (aclsAdminApiSupported) {
                KafkaUserAuthorizationSimple simple = (KafkaUserAuthorizationSimple) kafkaUser.getSpec().getAuthorization();
                result.setSimpleAclRules(simple.getAcls());
            } else {
                throw new InvalidResourceException("Simple authorization ACL rules are configured but not supported in the Kafka cluster configuration.");
            }
        }

        result.setQuotas(kafkaUser.getSpec().getQuotas());

        if (kafkaUser.getSpec().getTemplate() != null
                && kafkaUser.getSpec().getTemplate().getSecret() != null
                && kafkaUser.getSpec().getTemplate().getSecret().getMetadata() != null)  {
            result.templateSecretLabels = kafkaUser.getSpec().getTemplate().getSecret().getMetadata().getLabels();
            result.templateSecretAnnotations = kafkaUser.getSpec().getTemplate().getSecret().getMetadata().getAnnotations();
        }

        return result;
    }
@Test
    public void testFromCrdAclsWithAclsAdminApiSupportMissing()    {
        InvalidResourceException e = assertThrows(InvalidResourceException.class, () -> KafkaUserModel.fromCrd(scramShaUser, UserOperatorConfig.SECRET_PREFIX.defaultValue(), false));
        assertThat(e.getMessage(), is("Simple authorization ACL rules are configured but not supported in the Kafka cluster configuration."));
    }