@Test public void testCreateOrUpdateThrowsWhenCreateServiceThrows(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockBridgeOps = supplier.kafkaBridgeOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; String kbName = "foo"; String kbNamespace = "test"; Map<String, Object> metricsCm = new HashMap<>(); metricsCm.put("foo", "bar"); KafkaBridge kb = ResourceUtils.createKafkaBridge(kbNamespace, kbName, image, 1, BOOTSTRAP_SERVERS, KAFKA_BRIDGE_PRODUCER_SPEC, KAFKA_BRIDGE_CONSUMER_SPEC, KAFKA_BRIDGE_HTTP_SPEC, true); KafkaBridgeCluster bridge = KafkaBridgeCluster.fromCrd(kb, VERSIONS); kb.getSpec().setImage("some/different:image"); when(mockBridgeOps.get(kbNamespace, kbName)).thenReturn(kb); when(mockBridgeOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kb)); when(mockBridgeOps.updateStatusAsync(any(KafkaBridge.class))).thenReturn(Future.succeededFuture()); when(mockServiceOps.get(kbNamespace, bridge.getName())).thenReturn(bridge.generateService()); when(mockDcOps.get(kbNamespace, bridge.getName())).thenReturn(bridge.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> serviceNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> serviceNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Service> serviceCaptor = ArgumentCaptor.forClass(Service.class); when(mockServiceOps.reconcile(serviceNamespaceCaptor.capture(), serviceNameCaptor.capture(), serviceCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(dcNamespaceCaptor.capture(), dcNameCaptor.capture(), dcCaptor.capture())).thenReturn(Future.failedFuture("Failed")); ArgumentCaptor<String> dcScaleUpNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> dcScaleUpNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleUpReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleUp(dcScaleUpNamespaceCaptor.capture(), dcScaleUpNameCaptor.capture(), dcScaleUpReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcScaleDownNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> dcScaleDownNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleDownReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleDown(dcScaleDownNamespaceCaptor.capture(), dcScaleDownNameCaptor.capture(), dcScaleDownReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture()); when(mockBridgeOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaBridge()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); KafkaBridgeAssemblyOperator ops = new KafkaBridgeAssemblyOperator(vertx, new PlatformFeaturesAvailability(true, kubernetesVersion), new MockCertManager(), new PasswordGenerator(10, "a", "a"), supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS)); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaBridge.RESOURCE_KIND, kbNamespace, kbName), kb) .onComplete(context.failing(e -> async.flag())); }
@Override protected Future<KafkaBridgeStatus> createOrUpdate(Reconciliation reconciliation, KafkaBridge assemblyResource) { KafkaBridgeStatus kafkaBridgeStatus = new KafkaBridgeStatus(); String namespace = reconciliation.namespace(); KafkaBridgeCluster bridge; try { bridge = KafkaBridgeCluster.fromCrd(assemblyResource, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaBridgeStatus, e)); } ConfigMap logAndMetricsConfigMap = bridge.generateMetricsAndLogConfigMap(bridge.getLogging() instanceof ExternalLogging ? configMapOperations.get(namespace, ((ExternalLogging) bridge.getLogging()).getName()) : null); Promise<KafkaBridgeStatus> createOrUpdatePromise = Promise.promise(); boolean bridgeHasZeroReplicas = bridge.getReplicas() == 0; log.debug("{}: Updating Kafka Bridge cluster", reconciliation); kafkaBridgeServiceAccount(namespace, bridge) .compose(i -> deploymentOperations.scaleDown(namespace, bridge.getName(), bridge.getReplicas())) .compose(scale -> serviceOperations.reconcile(namespace, bridge.getServiceName(), bridge.generateService())) .compose(i -> configMapOperations.reconcile(namespace, bridge.getAncillaryConfigMapName(), logAndMetricsConfigMap)) .compose(i -> podDisruptionBudgetOperator.reconcile(namespace, bridge.getName(), bridge.generatePodDisruptionBudget())) .compose(i -> deploymentOperations.reconcile(namespace, bridge.getName(), bridge.generateDeployment(Collections.emptyMap(), pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> deploymentOperations.scaleUp(namespace, bridge.getName(), bridge.getReplicas())) .compose(i -> deploymentOperations.waitForObserved(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .compose(i -> bridgeHasZeroReplicas ? Future.succeededFuture() : deploymentOperations.readiness(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, reconciliationResult.mapEmpty()); if (!bridgeHasZeroReplicas) { int port = KafkaBridgeCluster.DEFAULT_REST_API_PORT; if (bridge.getHttp() != null) { port = bridge.getHttp().getPort(); } kafkaBridgeStatus.setUrl(KafkaBridgeResources.url(bridge.getCluster(), namespace, port)); } kafkaBridgeStatus.setReplicas(bridge.getReplicas()); kafkaBridgeStatus.setLabelSelector(bridge.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaBridgeStatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaBridgeStatus, reconciliationResult.cause())); } }); return createOrUpdatePromise.future(); }
KafkaBridgeAssemblyOperator extends AbstractAssemblyOperator<KubernetesClient, KafkaBridge, KafkaBridgeList, DoneableKafkaBridge, Resource<KafkaBridge, DoneableKafkaBridge>, KafkaBridgeSpec, KafkaBridgeStatus> { @Override protected Future<KafkaBridgeStatus> createOrUpdate(Reconciliation reconciliation, KafkaBridge assemblyResource) { KafkaBridgeStatus kafkaBridgeStatus = new KafkaBridgeStatus(); String namespace = reconciliation.namespace(); KafkaBridgeCluster bridge; try { bridge = KafkaBridgeCluster.fromCrd(assemblyResource, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaBridgeStatus, e)); } ConfigMap logAndMetricsConfigMap = bridge.generateMetricsAndLogConfigMap(bridge.getLogging() instanceof ExternalLogging ? configMapOperations.get(namespace, ((ExternalLogging) bridge.getLogging()).getName()) : null); Promise<KafkaBridgeStatus> createOrUpdatePromise = Promise.promise(); boolean bridgeHasZeroReplicas = bridge.getReplicas() == 0; log.debug("{}: Updating Kafka Bridge cluster", reconciliation); kafkaBridgeServiceAccount(namespace, bridge) .compose(i -> deploymentOperations.scaleDown(namespace, bridge.getName(), bridge.getReplicas())) .compose(scale -> serviceOperations.reconcile(namespace, bridge.getServiceName(), bridge.generateService())) .compose(i -> configMapOperations.reconcile(namespace, bridge.getAncillaryConfigMapName(), logAndMetricsConfigMap)) .compose(i -> podDisruptionBudgetOperator.reconcile(namespace, bridge.getName(), bridge.generatePodDisruptionBudget())) .compose(i -> deploymentOperations.reconcile(namespace, bridge.getName(), bridge.generateDeployment(Collections.emptyMap(), pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> deploymentOperations.scaleUp(namespace, bridge.getName(), bridge.getReplicas())) .compose(i -> deploymentOperations.waitForObserved(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .compose(i -> bridgeHasZeroReplicas ? Future.succeededFuture() : deploymentOperations.readiness(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, reconciliationResult.mapEmpty()); if (!bridgeHasZeroReplicas) { int port = KafkaBridgeCluster.DEFAULT_REST_API_PORT; if (bridge.getHttp() != null) { port = bridge.getHttp().getPort(); } kafkaBridgeStatus.setUrl(KafkaBridgeResources.url(bridge.getCluster(), namespace, port)); } kafkaBridgeStatus.setReplicas(bridge.getReplicas()); kafkaBridgeStatus.setLabelSelector(bridge.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaBridgeStatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaBridgeStatus, reconciliationResult.cause())); } }); return createOrUpdatePromise.future(); } }
KafkaBridgeAssemblyOperator extends AbstractAssemblyOperator<KubernetesClient, KafkaBridge, KafkaBridgeList, DoneableKafkaBridge, Resource<KafkaBridge, DoneableKafkaBridge>, KafkaBridgeSpec, KafkaBridgeStatus> { @Override protected Future<KafkaBridgeStatus> createOrUpdate(Reconciliation reconciliation, KafkaBridge assemblyResource) { KafkaBridgeStatus kafkaBridgeStatus = new KafkaBridgeStatus(); String namespace = reconciliation.namespace(); KafkaBridgeCluster bridge; try { bridge = KafkaBridgeCluster.fromCrd(assemblyResource, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaBridgeStatus, e)); } ConfigMap logAndMetricsConfigMap = bridge.generateMetricsAndLogConfigMap(bridge.getLogging() instanceof ExternalLogging ? configMapOperations.get(namespace, ((ExternalLogging) bridge.getLogging()).getName()) : null); Promise<KafkaBridgeStatus> createOrUpdatePromise = Promise.promise(); boolean bridgeHasZeroReplicas = bridge.getReplicas() == 0; log.debug("{}: Updating Kafka Bridge cluster", reconciliation); kafkaBridgeServiceAccount(namespace, bridge) .compose(i -> deploymentOperations.scaleDown(namespace, bridge.getName(), bridge.getReplicas())) .compose(scale -> serviceOperations.reconcile(namespace, bridge.getServiceName(), bridge.generateService())) .compose(i -> configMapOperations.reconcile(namespace, bridge.getAncillaryConfigMapName(), logAndMetricsConfigMap)) .compose(i -> podDisruptionBudgetOperator.reconcile(namespace, bridge.getName(), bridge.generatePodDisruptionBudget())) .compose(i -> deploymentOperations.reconcile(namespace, bridge.getName(), bridge.generateDeployment(Collections.emptyMap(), pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> deploymentOperations.scaleUp(namespace, bridge.getName(), bridge.getReplicas())) .compose(i -> deploymentOperations.waitForObserved(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .compose(i -> bridgeHasZeroReplicas ? Future.succeededFuture() : deploymentOperations.readiness(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, reconciliationResult.mapEmpty()); if (!bridgeHasZeroReplicas) { int port = KafkaBridgeCluster.DEFAULT_REST_API_PORT; if (bridge.getHttp() != null) { port = bridge.getHttp().getPort(); } kafkaBridgeStatus.setUrl(KafkaBridgeResources.url(bridge.getCluster(), namespace, port)); } kafkaBridgeStatus.setReplicas(bridge.getReplicas()); kafkaBridgeStatus.setLabelSelector(bridge.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaBridgeStatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaBridgeStatus, reconciliationResult.cause())); } }); return createOrUpdatePromise.future(); } KafkaBridgeAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                       CertManager certManager, PasswordGenerator passwordGenerator,
                                       ResourceOperatorSupplier supplier,
                                       ClusterOperatorConfig config); }
KafkaBridgeAssemblyOperator extends AbstractAssemblyOperator<KubernetesClient, KafkaBridge, KafkaBridgeList, DoneableKafkaBridge, Resource<KafkaBridge, DoneableKafkaBridge>, KafkaBridgeSpec, KafkaBridgeStatus> { @Override protected Future<KafkaBridgeStatus> createOrUpdate(Reconciliation reconciliation, KafkaBridge assemblyResource) { KafkaBridgeStatus kafkaBridgeStatus = new KafkaBridgeStatus(); String namespace = reconciliation.namespace(); KafkaBridgeCluster bridge; try { bridge = KafkaBridgeCluster.fromCrd(assemblyResource, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaBridgeStatus, e)); } ConfigMap logAndMetricsConfigMap = bridge.generateMetricsAndLogConfigMap(bridge.getLogging() instanceof ExternalLogging ? configMapOperations.get(namespace, ((ExternalLogging) bridge.getLogging()).getName()) : null); Promise<KafkaBridgeStatus> createOrUpdatePromise = Promise.promise(); boolean bridgeHasZeroReplicas = bridge.getReplicas() == 0; log.debug("{}: Updating Kafka Bridge cluster", reconciliation); kafkaBridgeServiceAccount(namespace, bridge) .compose(i -> deploymentOperations.scaleDown(namespace, bridge.getName(), bridge.getReplicas())) .compose(scale -> serviceOperations.reconcile(namespace, bridge.getServiceName(), bridge.generateService())) .compose(i -> configMapOperations.reconcile(namespace, bridge.getAncillaryConfigMapName(), logAndMetricsConfigMap)) .compose(i -> podDisruptionBudgetOperator.reconcile(namespace, bridge.getName(), bridge.generatePodDisruptionBudget())) .compose(i -> deploymentOperations.reconcile(namespace, bridge.getName(), bridge.generateDeployment(Collections.emptyMap(), pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> deploymentOperations.scaleUp(namespace, bridge.getName(), bridge.getReplicas())) .compose(i -> deploymentOperations.waitForObserved(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .compose(i -> bridgeHasZeroReplicas ? Future.succeededFuture() : deploymentOperations.readiness(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, reconciliationResult.mapEmpty()); if (!bridgeHasZeroReplicas) { int port = KafkaBridgeCluster.DEFAULT_REST_API_PORT; if (bridge.getHttp() != null) { port = bridge.getHttp().getPort(); } kafkaBridgeStatus.setUrl(KafkaBridgeResources.url(bridge.getCluster(), namespace, port)); } kafkaBridgeStatus.setReplicas(bridge.getReplicas()); kafkaBridgeStatus.setLabelSelector(bridge.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaBridgeStatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaBridgeStatus, reconciliationResult.cause())); } }); return createOrUpdatePromise.future(); } KafkaBridgeAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                       CertManager certManager, PasswordGenerator passwordGenerator,
                                       ResourceOperatorSupplier supplier,
                                       ClusterOperatorConfig config);  }
KafkaBridgeAssemblyOperator extends AbstractAssemblyOperator<KubernetesClient, KafkaBridge, KafkaBridgeList, DoneableKafkaBridge, Resource<KafkaBridge, DoneableKafkaBridge>, KafkaBridgeSpec, KafkaBridgeStatus> { @Override protected Future<KafkaBridgeStatus> createOrUpdate(Reconciliation reconciliation, KafkaBridge assemblyResource) { KafkaBridgeStatus kafkaBridgeStatus = new KafkaBridgeStatus(); String namespace = reconciliation.namespace(); KafkaBridgeCluster bridge; try { bridge = KafkaBridgeCluster.fromCrd(assemblyResource, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaBridgeStatus, e)); } ConfigMap logAndMetricsConfigMap = bridge.generateMetricsAndLogConfigMap(bridge.getLogging() instanceof ExternalLogging ? configMapOperations.get(namespace, ((ExternalLogging) bridge.getLogging()).getName()) : null); Promise<KafkaBridgeStatus> createOrUpdatePromise = Promise.promise(); boolean bridgeHasZeroReplicas = bridge.getReplicas() == 0; log.debug("{}: Updating Kafka Bridge cluster", reconciliation); kafkaBridgeServiceAccount(namespace, bridge) .compose(i -> deploymentOperations.scaleDown(namespace, bridge.getName(), bridge.getReplicas())) .compose(scale -> serviceOperations.reconcile(namespace, bridge.getServiceName(), bridge.generateService())) .compose(i -> configMapOperations.reconcile(namespace, bridge.getAncillaryConfigMapName(), logAndMetricsConfigMap)) .compose(i -> podDisruptionBudgetOperator.reconcile(namespace, bridge.getName(), bridge.generatePodDisruptionBudget())) .compose(i -> deploymentOperations.reconcile(namespace, bridge.getName(), bridge.generateDeployment(Collections.emptyMap(), pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> deploymentOperations.scaleUp(namespace, bridge.getName(), bridge.getReplicas())) .compose(i -> deploymentOperations.waitForObserved(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .compose(i -> bridgeHasZeroReplicas ? Future.succeededFuture() : deploymentOperations.readiness(namespace, bridge.getName(), 1_000, operationTimeoutMs)) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaBridgeStatus, reconciliationResult.mapEmpty()); if (!bridgeHasZeroReplicas) { int port = KafkaBridgeCluster.DEFAULT_REST_API_PORT; if (bridge.getHttp() != null) { port = bridge.getHttp().getPort(); } kafkaBridgeStatus.setUrl(KafkaBridgeResources.url(bridge.getCluster(), namespace, port)); } kafkaBridgeStatus.setReplicas(bridge.getReplicas()); kafkaBridgeStatus.setLabelSelector(bridge.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaBridgeStatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaBridgeStatus, reconciliationResult.cause())); } }); return createOrUpdatePromise.future(); } KafkaBridgeAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                       CertManager certManager, PasswordGenerator passwordGenerator,
                                       ResourceOperatorSupplier supplier,
                                       ClusterOperatorConfig config);   }