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);   }
@Test public void testCreateOrUpdateWithNoDiffCausesNoChanges(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); 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> serviceNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Service> serviceCaptor = ArgumentCaptor.forClass(Service.class); when(mockServiceOps.reconcile(eq(kbNamespace), serviceNameCaptor.capture(), serviceCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(eq(kbNamespace), dcNameCaptor.capture(), dcCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcScaleUpNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleUpReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleUp(eq(kbNamespace), dcScaleUpNameCaptor.capture(), dcScaleUpReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcScaleDownNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleDownReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleDown(eq(kbNamespace), dcScaleDownNameCaptor.capture(), dcScaleDownReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<PodDisruptionBudget> pdbCaptor = ArgumentCaptor.forClass(PodDisruptionBudget.class); when(mockPdbOps.reconcile(anyString(), any(), pdbCaptor.capture())).thenReturn(Future.succeededFuture()); 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.succeeding(v -> context.verify(() -> { List<Service> capturedServices = serviceCaptor.getAllValues(); assertThat(capturedServices, hasSize(1)); List<Deployment> capturedDc = dcCaptor.getAllValues(); assertThat(capturedDc, hasSize(1)); List<PodDisruptionBudget> capturedPdb = pdbCaptor.getAllValues(); assertThat(capturedPdb, hasSize(1)); PodDisruptionBudget pdb = capturedPdb.get(0); assertThat(pdb.getMetadata().getName(), is(bridge.getName())); assertThat(pdb, is(bridge.generatePodDisruptionBudget())); assertThat(dcScaleDownNameCaptor.getAllValues(), hasSize(1)); assertThat(dcScaleUpNameCaptor.getAllValues(), hasSize(1)); async.flag(); }))); }
@Test public void testCreateOrUpdateUpdatesCluster(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> metricsCmP = new HashMap<>(); metricsCmP.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> serviceNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Service> serviceCaptor = ArgumentCaptor.forClass(Service.class); when(mockServiceOps.reconcile(eq(kbNamespace), serviceNameCaptor.capture(), serviceCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(eq(kbNamespace), dcNameCaptor.capture(), dcCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcScaleUpNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleUpReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleUp(eq(kbNamespace), dcScaleUpNameCaptor.capture(), dcScaleUpReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcScaleDownNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleDownReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleDown(eq(kbNamespace), dcScaleDownNameCaptor.capture(), dcScaleDownReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<PodDisruptionBudget> pdbCaptor = ArgumentCaptor.forClass(PodDisruptionBudget.class); when(mockPdbOps.reconcile(anyString(), any(), pdbCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockBridgeOps.get(kbNamespace, kbName)).thenReturn(kb); ConfigMap metricsCm = new ConfigMapBuilder().withNewMetadata() .withName(KafkaBridgeResources.metricsAndLogConfigMapName(kbName)) .withNamespace(kbNamespace) .endMetadata() .withData(Collections.singletonMap(AbstractModel.ANCILLARY_CM_KEY_METRICS, METRICS_CONFIG)) .build(); when(mockCmOps.get(kbNamespace, KafkaBridgeResources.metricsAndLogConfigMapName(kbName))).thenReturn(metricsCm); ConfigMap loggingCm = new ConfigMapBuilder().withNewMetadata() .withName(KafkaBridgeResources.metricsAndLogConfigMapName(kbName)) .withNamespace(kbNamespace) .endMetadata() .withData(Collections.singletonMap(AbstractModel.ANCILLARY_CM_KEY_LOG_CONFIG, LOGGING_CONFIG)) .build(); when(mockCmOps.get(kbNamespace, KafkaBridgeResources.metricsAndLogConfigMapName(kbName))).thenReturn(metricsCm); Set<String> metricsCms = TestUtils.set(); doAnswer(invocation -> { metricsCms.add(invocation.getArgument(1)); return Future.succeededFuture(); }).when(mockCmOps).reconcile(eq(kbNamespace), anyString(), any()); 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.succeeding(v -> context.verify(() -> { KafkaBridgeCluster compareTo = KafkaBridgeCluster.fromCrd(kb, VERSIONS); List<Service> capturedServices = serviceCaptor.getAllValues(); assertThat(capturedServices, hasSize(1)); Service service = capturedServices.get(0); assertThat(service.getMetadata().getName(), is(compareTo.getServiceName())); assertThat(service, is(compareTo.generateService())); List<Deployment> capturedDc = dcCaptor.getAllValues(); assertThat(capturedDc, hasSize(1)); Deployment dc = capturedDc.get(0); assertThat(dc.getMetadata().getName(), is(compareTo.getName())); assertThat(dc, is(compareTo.generateDeployment(Collections.emptyMap(), true, null, null))); List<PodDisruptionBudget> capturedPdb = pdbCaptor.getAllValues(); assertThat(capturedPdb, hasSize(1)); PodDisruptionBudget pdb = capturedPdb.get(0); assertThat(pdb.getMetadata().getName(), is(compareTo.getName())); assertThat(pdb, is(compareTo.generatePodDisruptionBudget())); assertThat(dcScaleDownNameCaptor.getAllValues(), hasSize(1)); assertThat(dcScaleUpNameCaptor.getAllValues(), hasSize(1)); verify(mockCmOps, never()).createOrUpdate(any()); async.flag(); }))); }
@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())); }
@Test public void testCreateOrUpdateWithReplicasScaleUpToOne(VertxTestContext context) { final int scaleTo = 1; 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"; KafkaBridge kb = ResourceUtils.createEmptyKafkaBridge(kbNamespace, kbName); kb.getSpec().setReplicas(0); KafkaBridgeCluster bridge = KafkaBridgeCluster.fromCrd(kb, VERSIONS); kb.getSpec().setReplicas(scaleTo); 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()); Deployment dep = bridge.generateDeployment(new HashMap<>(), true, null, null); when(mockDcOps.get(kbNamespace, bridge.getName())).thenReturn(dep); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockServiceOps.reconcile(eq(kbNamespace), any(), any())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(eq(kbNamespace), any(), any())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kbNamespace, bridge.getName(), scaleTo); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kbNamespace, bridge.getName(), scaleTo); 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()))); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new PodDisruptionBudget()))); 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.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleUp(kbNamespace, bridge.getName(), scaleTo); async.flag(); }))); }
@Test public void testCreateOrUpdateWithScaleDown(VertxTestContext context) { int scaleTo = 1; 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"; KafkaBridge kb = ResourceUtils.createEmptyKafkaBridge(kbNamespace, kbName); kb.getSpec().setReplicas(scaleTo); KafkaBridge scaledDownCluster = new KafkaBridgeBuilder(kb) .editOrNewSpec() .withReplicas(scaleTo) .endSpec() .build(); KafkaBridgeCluster bridge = KafkaBridgeCluster.fromCrd(kb, VERSIONS); 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()); Deployment dep = bridge.generateDeployment(new HashMap<>(), true, null, null); when(mockDcOps.get(kbNamespace, bridge.getName())).thenReturn(dep); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockServiceOps.reconcile(eq(kbNamespace), any(), any())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(eq(kbNamespace), any(), any())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kbNamespace, bridge.getName(), scaleTo); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kbNamespace, bridge.getName(), scaleTo); 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()))); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new PodDisruptionBudget()))); 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), scaledDownCluster) .onComplete(context.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleUp(kbNamespace, bridge.getName(), scaleTo); verify(mockDcOps).scaleDown(kbNamespace, bridge.getName(), scaleTo); async.flag(); }))); }