KafkaMirrorMakerAssemblyOperator extends AbstractAssemblyOperator<KubernetesClient, KafkaMirrorMaker, KafkaMirrorMakerList, DoneableKafkaMirrorMaker, Resource<KafkaMirrorMaker, DoneableKafkaMirrorMaker>, KafkaMirrorMakerSpec, KafkaMirrorMakerStatus> { @Override protected Future<KafkaMirrorMakerStatus> createOrUpdate(Reconciliation reconciliation, KafkaMirrorMaker assemblyResource) { String namespace = reconciliation.namespace(); KafkaMirrorMakerCluster mirror; KafkaMirrorMakerStatus kafkaMirrorMakerStatus = new KafkaMirrorMakerStatus(); try { mirror = KafkaMirrorMakerCluster.fromCrd(assemblyResource, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaMirrorMakerStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaMirrorMakerStatus, e)); } ConfigMap logAndMetricsConfigMap = mirror.generateMetricsAndLogConfigMap(mirror.getLogging() instanceof ExternalLogging ? configMapOperations.get(namespace, ((ExternalLogging) mirror.getLogging()).getName()) : null); Map<String, String> annotations = new HashMap<>(1); annotations.put(Annotations.STRIMZI_LOGGING_ANNOTATION, logAndMetricsConfigMap.getData().get(mirror.ANCILLARY_CM_KEY_LOG_CONFIG)); Promise<KafkaMirrorMakerStatus> createOrUpdatePromise = Promise.promise(); boolean mirrorHasZeroReplicas = mirror.getReplicas() == 0; log.debug("{}: Updating Kafka Mirror Maker cluster", reconciliation); mirrorMakerServiceAccount(namespace, mirror) .compose(i -> deploymentOperations.scaleDown(namespace, mirror.getName(), mirror.getReplicas())) .compose(i -> configMapOperations.reconcile(namespace, mirror.getAncillaryConfigMapName(), logAndMetricsConfigMap)) .compose(i -> podDisruptionBudgetOperator.reconcile(namespace, mirror.getName(), mirror.generatePodDisruptionBudget())) .compose(i -> deploymentOperations.reconcile(namespace, mirror.getName(), mirror.generateDeployment(annotations, pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> deploymentOperations.scaleUp(namespace, mirror.getName(), mirror.getReplicas())) .compose(i -> deploymentOperations.waitForObserved(namespace, mirror.getName(), 1_000, operationTimeoutMs)) .compose(i -> mirrorHasZeroReplicas ? Future.succeededFuture() : deploymentOperations.readiness(namespace, mirror.getName(), 1_000, operationTimeoutMs)) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(assemblyResource, kafkaMirrorMakerStatus, reconciliationResult); kafkaMirrorMakerStatus.setReplicas(mirror.getReplicas()); kafkaMirrorMakerStatus.setLabelSelector(mirror.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaMirrorMakerStatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaMirrorMakerStatus, reconciliationResult.cause())); } } ); return createOrUpdatePromise.future(); } KafkaMirrorMakerAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                            CertManager certManager, PasswordGenerator passwordGenerator,
                                            ResourceOperatorSupplier supplier,
                                            ClusterOperatorConfig config);   }
@Test public void testUpdateClusterNoDiff(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockMirrorOps = supplier.mirrorMakerOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; String kmmName = "foo"; String kmmNamespace = "test"; KafkaMirrorMakerConsumerSpec consumer = new KafkaMirrorMakerConsumerSpecBuilder() .withBootstrapServers(consumerBootstrapServers) .withGroupId(groupId) .withNumStreams(numStreams) .build(); KafkaMirrorMakerProducerSpec producer = new KafkaMirrorMakerProducerSpecBuilder() .withBootstrapServers(producerBootstrapServers) .build(); Map<String, Object> metricsCm = new HashMap<>(); metricsCm.put("foo", "bar"); KafkaMirrorMaker kmm = ResourceUtils.createKafkaMirrorMaker(kmmNamespace, kmmName, image, producer, consumer, whitelist, metricsCm); KafkaMirrorMakerCluster mirror = KafkaMirrorMakerCluster.fromCrd(kmm, VERSIONS); when(mockMirrorOps.get(kmmNamespace, kmmName)).thenReturn(kmm); when(mockMirrorOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kmm)); when(mockMirrorOps.updateStatusAsync(any(KafkaMirrorMaker.class))).thenReturn(Future.succeededFuture()); when(mockDcOps.get(kmmNamespace, mirror.getName())).thenReturn(mirror.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(eq(kmmNamespace), 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(kmmNamespace), 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(kmmNamespace), dcScaleDownNameCaptor.capture(), dcScaleDownReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockDcOps.readiness(eq(kmmNamespace), eq(mirror.getName()), anyLong(), anyLong())).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()))); KafkaMirrorMakerAssemblyOperator ops = new KafkaMirrorMakerAssemblyOperator(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", KafkaMirrorMaker.RESOURCE_KIND, kmmNamespace, kmmName), kmm) .onComplete(context.succeeding(v -> context.verify(() -> { 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(mirror.getName())); assertThat("PodDisruptionBudgets are not equal", pdb, is(mirror.generatePodDisruptionBudget())); assertThat(dcScaleDownNameCaptor.getAllValues(), hasSize(1)); assertThat(dcScaleUpNameCaptor.getAllValues(), hasSize(1)); async.flag(); }))); }
@Test public void testUpdateCluster(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockMirrorOps = supplier.mirrorMakerOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; String kmmName = "foo"; String kmmNamespace = "test"; KafkaMirrorMakerConsumerSpec consumer = new KafkaMirrorMakerConsumerSpecBuilder() .withBootstrapServers(consumerBootstrapServers) .withGroupId(groupId) .withNumStreams(numStreams) .build(); KafkaMirrorMakerProducerSpec producer = new KafkaMirrorMakerProducerSpecBuilder() .withBootstrapServers(producerBootstrapServers) .build(); Map<String, Object> metricsCmP = new HashMap<>(); metricsCmP.put("foo", "bar"); KafkaMirrorMaker kmm = ResourceUtils.createKafkaMirrorMaker(kmmNamespace, kmmName, image, producer, consumer, whitelist, metricsCmP); KafkaMirrorMakerCluster mirror = KafkaMirrorMakerCluster.fromCrd(kmm, VERSIONS); kmm.getSpec().setImage("some/different:image"); when(mockMirrorOps.get(kmmNamespace, kmmName)).thenReturn(kmm); when(mockMirrorOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kmm)); when(mockMirrorOps.updateStatusAsync(any(KafkaMirrorMaker.class))).thenReturn(Future.succeededFuture()); when(mockDcOps.get(kmmNamespace, mirror.getName())).thenReturn(mirror.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(eq(kmmNamespace), eq(mirror.getName()), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(eq(kmmNamespace), 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(kmmNamespace), 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(kmmNamespace), 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(mockMirrorOps.get(kmmNamespace, kmmName)).thenReturn(kmm); ConfigMap metricsCm = new ConfigMapBuilder().withNewMetadata() .withName(KafkaMirrorMakerResources.metricsAndLogConfigMapName(kmmName)) .withNamespace(kmmNamespace) .endMetadata() .withData(Collections.singletonMap(AbstractModel.ANCILLARY_CM_KEY_METRICS, METRICS_CONFIG)) .build(); when(mockCmOps.get(kmmNamespace, KafkaMirrorMakerResources.metricsAndLogConfigMapName(kmmName))).thenReturn(metricsCm); ConfigMap loggingCm = new ConfigMapBuilder().withNewMetadata() .withName(KafkaMirrorMakerResources.metricsAndLogConfigMapName(kmmName)) .withNamespace(kmmNamespace) .endMetadata() .withData(Collections.singletonMap(AbstractModel.ANCILLARY_CM_KEY_LOG_CONFIG, LOGGING_CONFIG)) .build(); when(mockCmOps.get(kmmNamespace, KafkaMirrorMakerResources.metricsAndLogConfigMapName(kmmName))).thenReturn(metricsCm); Set<String> metricsCms = TestUtils.set(); doAnswer(invocation -> { metricsCms.add(invocation.getArgument(1)); return Future.succeededFuture(); }).when(mockCmOps).reconcile(eq(kmmNamespace), anyString(), any()); KafkaMirrorMakerAssemblyOperator ops = new KafkaMirrorMakerAssemblyOperator(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", KafkaMirrorMaker.RESOURCE_KIND, kmmNamespace, kmmName), kmm) .onComplete(context.succeeding(v -> context.verify(() -> { KafkaMirrorMakerCluster compareTo = KafkaMirrorMakerCluster.fromCrd(kmm, VERSIONS); List<Deployment> capturedDc = dcCaptor.getAllValues(); assertThat(capturedDc, hasSize(1)); Deployment dc = capturedDc.get(0); assertThat(dc.getMetadata().getName(), is(compareTo.getName())); Map<String, String> annotations = new HashMap(); annotations.put(Annotations.STRIMZI_LOGGING_ANNOTATION, loggingCm.getData().get(compareTo.ANCILLARY_CM_KEY_LOG_CONFIG)); assertThat("Deployments are not equal", dc, is(compareTo.generateDeployment(annotations, 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("PodDisruptionBudgets are not equal", pdb, is(compareTo.generatePodDisruptionBudget())); assertThat(dcScaleDownNameCaptor.getAllValues(), hasSize(1)); assertThat(dcScaleUpNameCaptor.getAllValues(), hasSize(1)); verify(mockCmOps, never()).createOrUpdate(any()); async.flag(); }))); }
@Test public void testUpdateClusterFailure(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockMirrorOps = supplier.mirrorMakerOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; String kmmName = "foo"; String kmmNamespace = "test"; KafkaMirrorMakerConsumerSpec consumer = new KafkaMirrorMakerConsumerSpecBuilder() .withBootstrapServers(consumerBootstrapServers) .withGroupId(groupId) .withNumStreams(numStreams) .build(); KafkaMirrorMakerProducerSpec producer = new KafkaMirrorMakerProducerSpecBuilder() .withBootstrapServers(producerBootstrapServers) .build(); Map<String, Object> metricsCm = new HashMap<>(); metricsCm.put("foo", "bar"); KafkaMirrorMaker kmm = ResourceUtils.createKafkaMirrorMaker(kmmNamespace, kmmName, image, producer, consumer, whitelist, metricsCm); KafkaMirrorMakerCluster mirror = KafkaMirrorMakerCluster.fromCrd(kmm, VERSIONS); kmm.getSpec().setImage("some/different:image"); when(mockMirrorOps.get(kmmNamespace, kmmName)).thenReturn(kmm); when(mockDcOps.get(kmmNamespace, mirror.getName())).thenReturn(mirror.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(eq(kmmNamespace), eq(mirror.getName()), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).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(mockMirrorOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaMirrorMaker()))); when(mockMirrorOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kmm)); when(mockMirrorOps.updateStatusAsync(any(KafkaMirrorMaker.class))).thenReturn(Future.succeededFuture()); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); KafkaMirrorMakerAssemblyOperator ops = new KafkaMirrorMakerAssemblyOperator(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", KafkaMirrorMaker.RESOURCE_KIND, kmmNamespace, kmmName), kmm) .onComplete(context.failing(v -> context.verify(() -> async.flag()))); }
@Test public void testUpdateClusterScaleUp(VertxTestContext context) { final int scaleTo = 4; ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockMirrorOps = supplier.mirrorMakerOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; String kmmName = "foo"; String kmmNamespace = "test"; KafkaMirrorMakerConsumerSpec consumer = new KafkaMirrorMakerConsumerSpecBuilder() .withBootstrapServers(consumerBootstrapServers) .withGroupId(groupId) .withNumStreams(numStreams) .build(); KafkaMirrorMakerProducerSpec producer = new KafkaMirrorMakerProducerSpecBuilder() .withBootstrapServers(producerBootstrapServers) .build(); Map<String, Object> metricsCm = new HashMap<>(); metricsCm.put("foo", "bar"); KafkaMirrorMaker kmm = ResourceUtils.createKafkaMirrorMaker(kmmNamespace, kmmName, image, producer, consumer, whitelist, metricsCm); KafkaMirrorMakerCluster mirror = KafkaMirrorMakerCluster.fromCrd(kmm, VERSIONS); kmm.getSpec().setReplicas(scaleTo); when(mockMirrorOps.get(kmmNamespace, kmmName)).thenReturn(kmm); when(mockDcOps.get(kmmNamespace, mirror.getName())).thenReturn(mirror.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(eq(kmmNamespace), eq(mirror.getName()), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(eq(kmmNamespace), any(), any())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kmmNamespace, mirror.getName(), scaleTo); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kmmNamespace, mirror.getName(), scaleTo); when(mockMirrorOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaMirrorMaker()))); when(mockMirrorOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kmm)); when(mockMirrorOps.updateStatusAsync(any(KafkaMirrorMaker.class))).thenReturn(Future.succeededFuture()); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture()); KafkaMirrorMakerAssemblyOperator ops = new KafkaMirrorMakerAssemblyOperator(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", KafkaMirrorMaker.RESOURCE_KIND, kmmNamespace, kmmName), kmm) .onComplete(context.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleUp(kmmNamespace, mirror.getName(), scaleTo); async.flag(); }))); }
@Test public void testUpdateClusterScaleDown(VertxTestContext context) { int scaleTo = 2; ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockMirrorOps = supplier.mirrorMakerOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; String kmmName = "foo"; String kmmNamespace = "test"; KafkaMirrorMakerConsumerSpec consumer = new KafkaMirrorMakerConsumerSpecBuilder() .withBootstrapServers(consumerBootstrapServers) .withGroupId(groupId) .withNumStreams(numStreams) .build(); KafkaMirrorMakerProducerSpec producer = new KafkaMirrorMakerProducerSpecBuilder() .withBootstrapServers(producerBootstrapServers) .build(); Map<String, Object> metricsCm = new HashMap<>(); metricsCm.put("foo", "bar"); KafkaMirrorMaker kmm = ResourceUtils.createKafkaMirrorMaker(kmmNamespace, kmmName, image, producer, consumer, whitelist, metricsCm); KafkaMirrorMakerCluster mirror = KafkaMirrorMakerCluster.fromCrd(kmm, VERSIONS); kmm.getSpec().setReplicas(scaleTo); when(mockMirrorOps.get(kmmNamespace, kmmName)).thenReturn(kmm); when(mockMirrorOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kmm)); when(mockMirrorOps.updateStatusAsync(any(KafkaMirrorMaker.class))).thenReturn(Future.succeededFuture()); when(mockDcOps.get(kmmNamespace, mirror.getName())).thenReturn(mirror.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(eq(kmmNamespace), eq(mirror.getName()), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(eq(kmmNamespace), any(), any())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kmmNamespace, mirror.getName(), scaleTo); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kmmNamespace, mirror.getName(), scaleTo); when(mockMirrorOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaMirrorMaker()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture()); KafkaMirrorMakerAssemblyOperator ops = new KafkaMirrorMakerAssemblyOperator(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", KafkaMirrorMaker.RESOURCE_KIND, kmmNamespace, kmmName), kmm) .onComplete(context.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleUp(kmmNamespace, mirror.getName(), scaleTo); async.flag(); }))); }