public static Future<Void> reconcileJmxSecret(Reconciliation reconciliation, SecretOperator secretOperator, SupportsJmx cluster)  {
        return secretOperator.getAsync(reconciliation.namespace(), cluster.jmx().secretName())
                .compose(currentJmxSecret -> {
                    Secret desiredJmxSecret = cluster.jmx().jmxSecret(currentJmxSecret);

                    if (desiredJmxSecret != null)  {
                        // Desired secret is not null => should be updated
                        return secretOperator.reconcile(reconciliation, reconciliation.namespace(), cluster.jmx().secretName(), desiredJmxSecret)
                                .map((Void) null);
                    } else if (currentJmxSecret != null)    {
                        // Desired secret is null but current is not => we should delete the secret
                        return secretOperator.reconcile(reconciliation, reconciliation.namespace(), cluster.jmx().secretName(), null)
                                .map((Void) null);
                    } else {
                        // Both current and desired secret are null => nothing to do
                        return Future.succeededFuture();
                    }
                });
    }
@Test
    public void testEnabledJmxWithAuthWithExistingSecret(VertxTestContext context) {
        KafkaClusterSpec spec = new KafkaClusterSpecBuilder()
                .withNewJmxOptions()
                    .withNewKafkaJmxAuthenticationPassword()
                    .endKafkaJmxAuthenticationPassword()
                .endJmxOptions()
                .build();
        JmxModel jmx = new JmxModel(NAMESPACE, NAME, LABELS, OWNER_REFERENCE, spec);

        SecretOperator mockSecretOps = mock(SecretOperator.class);
        when(mockSecretOps.getAsync(eq(NAMESPACE), eq(NAME))).thenReturn(Future.succeededFuture(EXISTING_JMX_SECRET));
        ArgumentCaptor<Secret> secretCaptor = ArgumentCaptor.forClass(Secret.class);
        when(mockSecretOps.reconcile(any(), any(), any(), secretCaptor.capture())).thenAnswer(i -> Future.succeededFuture(ReconcileResult.patched(i.getArgument(3))));

        Checkpoint async = context.checkpoint();
        ReconcilerUtils.reconcileJmxSecret(Reconciliation.DUMMY_RECONCILIATION, mockSecretOps, new MockJmxCluster(jmx))
                .onComplete(context.succeeding(v -> context.verify(() -> {
                    verify(mockSecretOps, times(1)).reconcile(eq(Reconciliation.DUMMY_RECONCILIATION), eq(NAMESPACE), eq(NAME), any());

                    Secret secret = secretCaptor.getValue();
                    assertThat(secret, is(notNullValue()));
                    assertThat(secret.getMetadata().getName(), is(NAME));
                    assertThat(secret.getMetadata().getNamespace(), is(NAMESPACE));
                    assertThat(secret.getMetadata().getOwnerReferences(), is(List.of(OWNER_REFERENCE)));
                    assertThat(secret.getMetadata().getLabels(), is(LABELS.toMap()));
                    assertThat(secret.getMetadata().getAnnotations(), is(nullValue()));
                    assertThat(secret.getData().size(), is(2));
                    assertThat(secret.getData().get("jmx-username"), is("username"));
                    assertThat(secret.getData().get("jmx-password"), is("password"));

                    async.flag();
                })));
    }