KafkaUserOperator extends AbstractOperator<KafkaUser, KafkaUserSpec, KafkaUserStatus,
        CrdOperator<KubernetesClient, KafkaUser, KafkaUserList, DoneableKafkaUser>> { @Override protected Future<Boolean> delete(Reconciliation reconciliation) { String namespace = reconciliation.namespace(); String user = reconciliation.name(); log.debug("{}: Deleting User", reconciliation, user, namespace); return CompositeFuture.join(secretOperations.reconcile(namespace, KafkaUserModel.getSecretName(user), null), aclOperations.reconcile(KafkaUserModel.getTlsUserName(user), null), aclOperations.reconcile(KafkaUserModel.getScramUserName(user), null), scramShaCredentialOperator.reconcile(KafkaUserModel.getScramUserName(user), null) .compose(ignore -> kafkaUserQuotasOperator.reconcile(KafkaUserModel.getTlsUserName(user), null)) .compose(ignore -> kafkaUserQuotasOperator.reconcile(KafkaUserModel.getScramUserName(user), null))) .map(Boolean.TRUE); } KafkaUserOperator(Vertx vertx,
                             CertManager certManager,
                             CrdOperator<KubernetesClient, KafkaUser, KafkaUserList, DoneableKafkaUser> crdOperator,
                             Labels labels,
                             SecretOperator secretOperations,
                             ScramShaCredentialsOperator scramShaCredentialOperator,
                             KafkaUserQuotasOperator kafkaUserQuotasOperator,
                             SimpleAclOperator aclOperations, String caCertName, String caKeyName, String caNamespace); @Override Optional<LabelSelector> selector(); @Override Future<Set<NamespaceAndName>> allResourceNames(String namespace);  }
@Test public void testDeleteTlsUser(VertxTestContext context) { CrdOperator mockCrdOps = mock(CrdOperator.class); SecretOperator mockSecretOps = mock(SecretOperator.class); SimpleAclOperator aclOps = mock(SimpleAclOperator.class); ScramShaCredentialsOperator scramOps = mock(ScramShaCredentialsOperator.class); KafkaUserQuotasOperator quotasOps = mock(KafkaUserQuotasOperator.class); ArgumentCaptor<String> secretNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> secretNameCaptor = ArgumentCaptor.forClass(String.class); when(mockSecretOps.reconcile(secretNamespaceCaptor.capture(), secretNameCaptor.capture(), isNull())).thenReturn(Future.succeededFuture()); when(scramOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> aclNameCaptor = ArgumentCaptor.forClass(String.class); when(aclOps.reconcile(aclNameCaptor.capture(), isNull())).thenReturn(Future.succeededFuture()); when(quotasOps.reconcile(anyString(), eq(null))).thenReturn(Future.succeededFuture()); KafkaUserOperator op = new KafkaUserOperator(vertx, mockCertManager, mockCrdOps, Labels.EMPTY, mockSecretOps, scramOps, quotasOps, aclOps, ResourceUtils.CA_CERT_NAME, ResourceUtils.CA_KEY_NAME, ResourceUtils.NAMESPACE); Checkpoint async = context.checkpoint(); op.delete(new Reconciliation("test-trigger", KafkaUser.RESOURCE_KIND, ResourceUtils.NAMESPACE, ResourceUtils.NAME)) .onComplete(context.succeeding(v -> context.verify(() -> { List<String> capturedNames = secretNameCaptor.getAllValues(); assertThat(capturedNames, hasSize(1)); assertThat(capturedNames.get(0), is(ResourceUtils.NAME)); List<String> capturedNamespaces = secretNamespaceCaptor.getAllValues(); assertThat(capturedNamespaces, hasSize(1)); assertThat(capturedNamespaces.get(0), is(ResourceUtils.NAMESPACE)); List<String> capturedAclNames = aclNameCaptor.getAllValues(); assertThat(capturedAclNames, hasSize(2)); assertThat(capturedAclNames.get(0), is(KafkaUserModel.getTlsUserName(ResourceUtils.NAME))); assertThat(capturedAclNames.get(1), is(KafkaUserModel.getScramUserName(ResourceUtils.NAME))); async.flag(); }))); }