public static Future<Integer> authTlsHash(SecretOperator secretOperations, String namespace, KafkaClientAuthentication auth, List<CertSecretSource> certSecretSources) {
        Future<Integer> tlsFuture;
        if (certSecretSources == null || certSecretSources.isEmpty()) {
            tlsFuture = Future.succeededFuture(0);
        } else {
            // get all TLS trusted certs, compute hash from each of them, sum hashes
            tlsFuture = Future.join(certSecretSources.stream().map(certSecretSource ->
                    getCertificateAsync(secretOperations, namespace, certSecretSource)
                    .compose(cert -> Future.succeededFuture(cert.hashCode()))).collect(Collectors.toList()))
                .compose(hashes -> Future.succeededFuture(hashes.list().stream().mapToInt(e -> (int) e).sum()));
        }

        if (auth == null) {
            return tlsFuture;
        } else {
            // compute hash from Auth
            if (auth instanceof KafkaClientAuthenticationScram) {
                // only passwordSecret can be changed
                return tlsFuture.compose(tlsHash -> getPasswordAsync(secretOperations, namespace, auth)
                        .compose(password -> Future.succeededFuture(password.hashCode() + tlsHash)));
            } else if (auth instanceof KafkaClientAuthenticationPlain) {
                // only passwordSecret can be changed
                return tlsFuture.compose(tlsHash -> getPasswordAsync(secretOperations, namespace, auth)
                        .compose(password -> Future.succeededFuture(password.hashCode() + tlsHash)));
            } else if (auth instanceof KafkaClientAuthenticationTls) {
                // custom cert can be used (and changed)
                return ((KafkaClientAuthenticationTls) auth).getCertificateAndKey() == null ? tlsFuture :
                        tlsFuture.compose(tlsHash -> getCertificateAndKeyAsync(secretOperations, namespace, (KafkaClientAuthenticationTls) auth)
                        .compose(crtAndKey -> Future.succeededFuture(crtAndKey.certAsBase64String().hashCode() + crtAndKey.keyAsBase64String().hashCode() + tlsHash)));
            } else if (auth instanceof KafkaClientAuthenticationOAuth) {
                List<Future<Integer>> futureList = ((KafkaClientAuthenticationOAuth) auth).getTlsTrustedCertificates() == null ?
                        new ArrayList<>() : ((KafkaClientAuthenticationOAuth) auth).getTlsTrustedCertificates().stream().map(certSecretSource ->
                        getCertificateAsync(secretOperations, namespace, certSecretSource)
                                .compose(cert -> Future.succeededFuture(cert.hashCode()))).collect(Collectors.toList());
                futureList.add(tlsFuture);
                futureList.add(addSecretHash(secretOperations, namespace, ((KafkaClientAuthenticationOAuth) auth).getAccessToken()));
                futureList.add(addSecretHash(secretOperations, namespace, ((KafkaClientAuthenticationOAuth) auth).getClientSecret()));
                futureList.add(addSecretHash(secretOperations, namespace, ((KafkaClientAuthenticationOAuth) auth).getRefreshToken()));
                return Future.join(futureList)
                        .compose(hashes -> Future.succeededFuture(hashes.list().stream().mapToInt(e -> (int) e).sum()));
            } else {
                // unknown Auth type
                return tlsFuture;
            }
        }
    }
@Test
    void getHashFailure() {
        String namespace = "ns";

        GenericSecretSource at = new GenericSecretSourceBuilder()
                .withSecretName("top-secret-at")
                .withKey("key")
                .build();

        GenericSecretSource cs = new GenericSecretSourceBuilder()
                .withSecretName("top-secret-cs")
                .withKey("key")
                .build();

        GenericSecretSource rt = new GenericSecretSourceBuilder()
                .withSecretName("top-secret-rt")
                .withKey("key")
                .build();
        KafkaClientAuthentication kcu = new KafkaClientAuthenticationOAuthBuilder()
                .withAccessToken(at)
                .withRefreshToken(rt)
                .withClientSecret(cs)
                .build();

        CertSecretSource css = new CertSecretSourceBuilder()
                .withCertificate("key")
                .withSecretName("css-secret")
                .build();

        Secret secret = new SecretBuilder()
                .withData(Map.of("key", "value"))
                .build();

        SecretOperator secretOps = mock(SecretOperator.class);
        when(secretOps.getAsync(eq(namespace), eq("top-secret-at"))).thenReturn(Future.succeededFuture(secret));
        when(secretOps.getAsync(eq(namespace), eq("top-secret-rt"))).thenReturn(Future.succeededFuture(secret));
        when(secretOps.getAsync(eq(namespace), eq("top-secret-cs"))).thenReturn(Future.succeededFuture(null));
        when(secretOps.getAsync(eq(namespace), eq("css-secret"))).thenReturn(Future.succeededFuture(secret));
        Future<Integer> res = VertxUtil.authTlsHash(secretOps, "ns", kcu, singletonList(css));
        res.onComplete(v -> {
            assertThat(v.succeeded(), is(false));
            assertThat(v.cause().getMessage(), is("Secret top-secret-cs not found"));
        });
    }