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 testFromCrdScramShaUserWithMissingPasswordKeyThrows()    {
        KafkaUser missingKey = new KafkaUserBuilder(scramShaUser)
                .editSpec()
                    .withNewKafkaUserScramSha512ClientAuthentication()
                        .withNewPassword()
                            .withNewValueFrom()
                                .withNewSecretKeyRef(null, "my-secret", false)
                            .endValueFrom()
                        .endPassword()
                    .endKafkaUserScramSha512ClientAuthentication()
                .endSpec()
                .build();

        InvalidResourceException e = assertThrows(InvalidResourceException.class, () -> {
            KafkaUserModel.fromCrd(missingKey, UserOperatorConfig.SECRET_PREFIX.defaultValue(), Boolean.parseBoolean(UserOperatorConfig.ACLS_ADMIN_API_SUPPORTED.defaultValue()));
        });

        assertThat(e.getMessage(), is("Resource requests custom SCRAM-SHA-512 password but doesn't specify the secret name and/or key"));
    }