KafkaConnectS2IAssemblyOperator extends AbstractConnectOperator<OpenShiftClient, KafkaConnectS2I, KafkaConnectS2IList, DoneableKafkaConnectS2I, Resource<KafkaConnectS2I, DoneableKafkaConnectS2I>, KafkaConnectS2ISpec, KafkaConnectS2IStatus> { @Override public Future<KafkaConnectS2IStatus> createOrUpdate(Reconciliation reconciliation, KafkaConnectS2I kafkaConnectS2I) { KafkaConnectS2ICluster connect; KafkaConnectS2IStatus kafkaConnectS2Istatus = new KafkaConnectS2IStatus(); try { connect = KafkaConnectS2ICluster.fromCrd(kafkaConnectS2I, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(kafkaConnectS2I, kafkaConnectS2Istatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaConnectS2Istatus, e)); } Promise<KafkaConnectS2IStatus> createOrUpdatePromise = Promise.promise(); String namespace = reconciliation.namespace(); connect.generateBuildConfig(); ConfigMap logAndMetricsConfigMap = connect.generateMetricsAndLogConfigMap(connect.getLogging() instanceof ExternalLogging ? configMapOperations.get(namespace, ((ExternalLogging) connect.getLogging()).getName()) : null); Map<String, String> annotations = new HashMap<>(1); annotations.put(Annotations.ANNO_STRIMZI_LOGGING_DYNAMICALLY_UNCHANGEABLE_HASH, Util.stringHash(Util.getLoggingDynamicallyUnmodifiableEntries(logAndMetricsConfigMap.getData().get(AbstractModel.ANCILLARY_CM_KEY_LOG_CONFIG)))); String desiredLogging = logAndMetricsConfigMap.getData().get(AbstractModel.ANCILLARY_CM_KEY_LOG_CONFIG); boolean connectHasZeroReplicas = connect.getReplicas() == 0; log.debug("{}: Updating Kafka Connect S2I cluster", reconciliation); connectOperations.getAsync(kafkaConnectS2I.getMetadata().getNamespace(), kafkaConnectS2I.getMetadata().getName()) .compose(otherConnect -> { if (otherConnect != null && kafkaConnectS2I.getMetadata().getCreationTimestamp().compareTo(otherConnect.getMetadata().getCreationTimestamp()) >= 0) { return Future.failedFuture("Both KafkaConnect and KafkaConnectS2I exist with the same name. " + "KafkaConnect is older and will be used while this custom resource will be ignored."); } else { return Future.succeededFuture(); } }) .compose(i -> { if (pfa.hasImages() && pfa.hasApps() && pfa.hasBuilds()) { return Future.succeededFuture(); } else { return Future.failedFuture("The OpenShift build, image or apps APIs are not available in this Kubernetes cluster. " + "Kafka Connect S2I deployment cannot be enabled."); } }) .compose(i -> connectServiceAccount(namespace, connect)) .compose(i -> networkPolicyOperator.reconcile(namespace, connect.getName(), connect.generateNetworkPolicy(pfa.isNamespaceAndPodSelectorNetworkPolicySupported(), isUseResources(kafkaConnectS2I)))) .compose(i -> deploymentConfigOperations.scaleDown(namespace, connect.getName(), connect.getReplicas())) .compose(scale -> serviceOperations.reconcile(namespace, connect.getServiceName(), connect.generateService())) .compose(i -> configMapOperations.reconcile(namespace, connect.getAncillaryConfigMapName(), logAndMetricsConfigMap)) .compose(i -> deploymentConfigOperations.reconcile(namespace, connect.getName(), connect.generateDeploymentConfig(annotations, pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> imagesStreamOperations.reconcile(namespace, KafkaConnectS2IResources.sourceImageStreamName(connect.getCluster()), connect.generateSourceImageStream())) .compose(i -> imagesStreamOperations.reconcile(namespace, KafkaConnectS2IResources.targetImageStreamName(connect.getCluster()), connect.generateTargetImageStream())) .compose(i -> podDisruptionBudgetOperator.reconcile(namespace, connect.getName(), connect.generatePodDisruptionBudget())) .compose(i -> buildConfigOperations.reconcile(namespace, KafkaConnectS2IResources.buildConfigName(connect.getCluster()), connect.generateBuildConfig())) .compose(i -> deploymentConfigOperations.scaleUp(namespace, connect.getName(), connect.getReplicas())) .compose(i -> deploymentConfigOperations.waitForObserved(namespace, connect.getName(), 1_000, operationTimeoutMs)) .compose(i -> connectHasZeroReplicas ? Future.succeededFuture() : deploymentConfigOperations.readiness(namespace, connect.getName(), 1_000, operationTimeoutMs)) .compose(i -> reconcileConnectors(reconciliation, kafkaConnectS2I, kafkaConnectS2Istatus, connectHasZeroReplicas, desiredLogging, connect.getDefaultLogConfig())) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(kafkaConnectS2I, kafkaConnectS2Istatus, reconciliationResult); if (!connectHasZeroReplicas) { kafkaConnectS2Istatus.setUrl(KafkaConnectS2IResources.url(connect.getCluster(), namespace, KafkaConnectS2ICluster.REST_API_PORT)); kafkaConnectS2Istatus.setBuildConfigName(KafkaConnectS2IResources.buildConfigName(connect.getCluster())); } kafkaConnectS2Istatus.setReplicas(connect.getReplicas()); kafkaConnectS2Istatus.setLabelSelector(connect.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaConnectS2Istatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaConnectS2Istatus, reconciliationResult.cause())); } }); return createOrUpdatePromise.future(); } @SuppressWarnings("checkstyle:parameternumber")  KafkaConnectS2IAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,

                                           ResourceOperatorSupplier supplier,
                                           ClusterOperatorConfig config);  KafkaConnectS2IAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                           ResourceOperatorSupplier supplier,
                                           ClusterOperatorConfig config,
                                           Function<Vertx, KafkaConnectApi> connectClientProvider); @Override Future<KafkaConnectS2IStatus> createOrUpdate(Reconciliation reconciliation, KafkaConnectS2I kafkaConnectS2I);  }
@Test public void testCreateOrUpdateDoesNotUpdateWithNoDiff(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; CrdOperator mockConnectOps = supplier.connectOperator; DeploymentConfigOperator mockDcOps = supplier.deploymentConfigOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; BuildConfigOperator mockBcOps = supplier.buildConfigOperations; ImageStreamOperator mockIsOps = supplier.imagesStreamOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcs2iName = "foo"; String kcs2iNamespace = "test"; KafkaConnectS2I kcs2i = ResourceUtils.createEmptyKafkaConnectS2I(kcs2iNamespace, kcs2iName); KafkaConnectS2ICluster connect = KafkaConnectS2ICluster.fromCrd(kcs2i, VERSIONS); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectS2IOps.get(kcs2iNamespace, kcs2iName)).thenReturn(kcs2i); when(mockConnectS2IOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kcs2i)); when(mockConnectS2IOps.updateStatusAsync(any(KafkaConnectS2I.class))).thenReturn(Future.succeededFuture()); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(null)); when(mockServiceOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateDeploymentConfig(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()); when(mockIsOps.get(kcs2iNamespace, KafkaConnectS2IResources.sourceImageStreamName(connect.getCluster()))).thenReturn(connect.generateSourceImageStream()); when(mockIsOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateTargetImageStream()); when(mockBcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateBuildConfig()); when(mockPdbOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generatePodDisruptionBudget()); 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<DeploymentConfig> dcCaptor = ArgumentCaptor.forClass(DeploymentConfig.class); when(mockDcOps.reconcile(dcNamespaceCaptor.capture(), dcNameCaptor.capture(), dcCaptor.capture())).thenReturn(Future.succeededFuture()); 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()); ArgumentCaptor<String> isNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> isNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<ImageStream> isCaptor = ArgumentCaptor.forClass(ImageStream.class); when(mockIsOps.reconcile(isNamespaceCaptor.capture(), isNameCaptor.capture(), isCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> bcNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> bcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<BuildConfig> bcCaptor = ArgumentCaptor.forClass(BuildConfig.class); when(mockBcOps.reconcile(bcNamespaceCaptor.capture(), bcNameCaptor.capture(), bcCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> pdbNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> pdbNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<PodDisruptionBudget> pdbCaptor = ArgumentCaptor.forClass(PodDisruptionBudget.class); when(mockPdbOps.reconcile(pdbNamespaceCaptor.capture(), pdbNameCaptor.capture(), pdbCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnectS2I()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockNetPolOps.reconcile(eq(kcs2i.getMetadata().getNamespace()), eq(KafkaConnectS2IResources.deploymentName(kcs2i.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); KafkaConnectApi mockConnectClient = mock(KafkaConnectApi.class); when(mockConnectClient.list(anyString(), anyInt())).thenReturn(Future.succeededFuture(emptyList())); ConnectorPlugin plugin1 = new ConnectorPluginBuilder() .withConnectorClass("io.strimzi.MyClass") .withType("sink") .withVersion("1.0.0") .build(); when(mockConnectClient.listConnectorPlugins(anyString(), anyInt())).thenReturn(Future.succeededFuture(singletonList(plugin1))); when(mockConnectClient.updateConnectLoggers(anyString(), anyInt(), anyString(), any(OrderedProperties.class))).thenReturn(Future.succeededFuture()); PlatformFeaturesAvailability pfa = new PlatformFeaturesAvailability(true, kubernetesVersion); KafkaConnectS2IAssemblyOperator ops = new KafkaConnectS2IAssemblyOperator(vertx, pfa, supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnectS2I.RESOURCE_KIND, kcs2iNamespace, kcs2iName), kcs2i) .onComplete(context.succeeding(v -> context.verify(() -> { List<Service> capturedServices = serviceCaptor.getAllValues(); assertThat(capturedServices, hasSize(1)); List<DeploymentConfig> capturedDc = dcCaptor.getAllValues(); assertThat(capturedDc, hasSize(1)); List<BuildConfig> capturedBc = bcCaptor.getAllValues(); assertThat(capturedBc, hasSize(1)); List<PodDisruptionBudget> capturedPdb = pdbCaptor.getAllValues(); assertThat(capturedPdb, hasSize(1)); List<ImageStream> capturedIs = isCaptor.getAllValues(); assertThat(capturedIs, hasSize(2)); assertThat(dcScaleDownNameCaptor.getAllValues(), hasSize(1)); assertThat(dcScaleUpNameCaptor.getAllValues(), hasSize(1)); async.flag(); }))); }
@SuppressWarnings({"checkstyle:JavaNCSS", "checkstyle:MethodLength"}) @Test public void testUpdateCluster(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; CrdOperator mockConnectOps = supplier.connectOperator; DeploymentConfigOperator mockDcOps = supplier.deploymentConfigOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; BuildConfigOperator mockBcOps = supplier.buildConfigOperations; ImageStreamOperator mockIsOps = supplier.imagesStreamOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcs2iName = "foo"; String kcs2iNamespace = "test"; KafkaConnectS2I kcs2i = ResourceUtils.createEmptyKafkaConnectS2I(kcs2iNamespace, kcs2iName); KafkaConnectS2ICluster connect = KafkaConnectS2ICluster.fromCrd(kcs2i, VERSIONS); kcs2i.getSpec().setImage("some/different:image"); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectS2IOps.get(kcs2iNamespace, kcs2iName)).thenReturn(kcs2i); when(mockConnectS2IOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kcs2i)); when(mockConnectS2IOps.updateStatusAsync(any(KafkaConnectS2I.class))).thenReturn(Future.succeededFuture()); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(null)); when(mockServiceOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateDeploymentConfig(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()); when(mockIsOps.get(kcs2iNamespace, KafkaConnectS2IResources.sourceImageStreamName(connect.getCluster()))).thenReturn(connect.generateSourceImageStream()); when(mockIsOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateTargetImageStream()); when(mockBcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateBuildConfig()); when(mockPdbOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generatePodDisruptionBudget()); 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<DeploymentConfig> dcCaptor = ArgumentCaptor.forClass(DeploymentConfig.class); when(mockDcOps.reconcile(dcNamespaceCaptor.capture(), dcNameCaptor.capture(), dcCaptor.capture())).thenReturn(Future.succeededFuture()); 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()); ArgumentCaptor<String> isNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> isNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<ImageStream> isCaptor = ArgumentCaptor.forClass(ImageStream.class); when(mockIsOps.reconcile(isNamespaceCaptor.capture(), isNameCaptor.capture(), isCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> bcNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> bcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<BuildConfig> bcCaptor = ArgumentCaptor.forClass(BuildConfig.class); when(mockBcOps.reconcile(bcNamespaceCaptor.capture(), bcNameCaptor.capture(), bcCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockNetPolOps.reconcile(eq(kcs2i.getMetadata().getNamespace()), eq(KafkaConnectS2IResources.deploymentName(kcs2i.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); when(mockConnectS2IOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnectS2I()))); ArgumentCaptor<String> pdbNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> pdbNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<PodDisruptionBudget> pdbCaptor = ArgumentCaptor.forClass(PodDisruptionBudget.class); when(mockPdbOps.reconcile(pdbNamespaceCaptor.capture(), pdbNameCaptor.capture(), pdbCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.get(kcs2iNamespace, kcs2iName)).thenReturn(kcs2i); ConfigMap metricsCm = new ConfigMapBuilder().withNewMetadata() .withName(KafkaConnectS2IResources.metricsAndLogConfigMapName(kcs2iName)) .withNamespace(kcs2iNamespace) .endMetadata() .withData(Collections.singletonMap(AbstractModel.ANCILLARY_CM_KEY_METRICS, METRICS_CONFIG)) .build(); when(mockCmOps.get(kcs2iNamespace, KafkaConnectS2IResources.metricsAndLogConfigMapName(kcs2iName))).thenReturn(metricsCm); Set<String> metricsCms = TestUtils.set(); doAnswer(invocation -> { metricsCms.add(invocation.getArgument(1)); return Future.succeededFuture(); }).when(mockCmOps).reconcile(eq(kcs2iNamespace), anyString(), any()); KafkaConnectApi mockConnectClient = mock(KafkaConnectApi.class); when(mockConnectClient.list(anyString(), anyInt())).thenReturn(Future.succeededFuture(emptyList())); ConnectorPlugin plugin1 = new ConnectorPluginBuilder() .withConnectorClass("io.strimzi.MyClass") .withType("sink") .withVersion("1.0.0") .build(); when(mockConnectClient.listConnectorPlugins(anyString(), anyInt())).thenReturn(Future.succeededFuture(singletonList(plugin1))); when(mockConnectClient.updateConnectLoggers(anyString(), anyInt(), anyString(), any(OrderedProperties.class))).thenReturn(Future.succeededFuture()); PlatformFeaturesAvailability pfa = new PlatformFeaturesAvailability(true, kubernetesVersion); KafkaConnectS2IAssemblyOperator ops = new KafkaConnectS2IAssemblyOperator(vertx, pfa, supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnectS2I.RESOURCE_KIND, kcs2iNamespace, kcs2iName), kcs2i) .onComplete(context.succeeding(v -> context.verify(() -> { KafkaConnectS2ICluster compareTo = KafkaConnectS2ICluster.fromCrd(kcs2i, 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<DeploymentConfig> capturedDc = dcCaptor.getAllValues(); assertThat(capturedDc, hasSize(1)); DeploymentConfig dc = capturedDc.get(0); assertThat(dc.getMetadata().getName(), is(compareTo.getName())); Map annotations = new HashMap(); annotations.put(Annotations.ANNO_STRIMZI_LOGGING_DYNAMICALLY_UNCHANGEABLE_HASH, Util.stringHash(Util.getLoggingDynamicallyUnmodifiableEntries(LOGGING_CONFIG))); assertThat(dc, is(compareTo.generateDeploymentConfig(annotations, true, null, null))); List<BuildConfig> capturedBc = bcCaptor.getAllValues(); assertThat(capturedBc, hasSize(1)); BuildConfig bc = capturedBc.get(0); assertThat(bc.getMetadata().getName(), is(compareTo.getName())); assertThat(bc, is(compareTo.generateBuildConfig())); 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())); List<ImageStream> capturedIs = isCaptor.getAllValues(); assertThat(capturedIs, hasSize(2)); int sourceImageStreamIndex = (KafkaConnectS2IResources.sourceImageStreamName(compareTo.getCluster())).equals(capturedIs.get(0).getMetadata().getName()) ? 0 : 1; int targetImageStreamIndex = (compareTo.getName()).equals(capturedIs.get(0).getMetadata().getName()) ? 0 : 1; ImageStream source = capturedIs.get(sourceImageStreamIndex); assertThat(source.getMetadata().getName(), is(KafkaConnectS2IResources.sourceImageStreamName(compareTo.getCluster()))); assertThat(source, is(compareTo.generateSourceImageStream())); ImageStream target = capturedIs.get(targetImageStreamIndex); assertThat(target.getMetadata().getName(), is(compareTo.getName())); assertThat(target, is(compareTo.generateTargetImageStream())); async.flag(); }))); }
@Test public void testCreateOrUpdateFailsWhenDeploymentUpdateFails(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; CrdOperator mockConnectOps = supplier.connectOperator; DeploymentConfigOperator mockDcOps = supplier.deploymentConfigOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; BuildConfigOperator mockBcOps = supplier.buildConfigOperations; ImageStreamOperator mockIsOps = supplier.imagesStreamOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; String kcs2iName = "foo"; String kcs2iNamespace = "test"; KafkaConnectS2I kcs2i = ResourceUtils.createEmptyKafkaConnectS2I(kcs2iNamespace, kcs2iName); KafkaConnectS2ICluster connect = KafkaConnectS2ICluster.fromCrd(kcs2i, VERSIONS); kcs2i.getSpec().setImage("some/different:image"); when(mockConnectS2IOps.get(kcs2iNamespace, kcs2iName)).thenReturn(kcs2i); when(mockServiceOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateDeploymentConfig(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()); when(mockIsOps.get(kcs2iNamespace, KafkaConnectS2IResources.sourceImageStreamName(connect.getCluster()))).thenReturn(connect.generateSourceImageStream()); when(mockIsOps.get(kcs2iNamespace, KafkaConnectS2IResources.targetImageStreamName(connect.getCluster()))).thenReturn(connect.generateTargetImageStream()); when(mockBcOps.get(kcs2iNamespace, KafkaConnectS2IResources.buildConfigName(connect.getCluster()))).thenReturn(connect.generateBuildConfig()); when(mockPdbOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generatePodDisruptionBudget()); when(mockConnectOps.getAsync(kcs2iNamespace, kcs2iName)).thenReturn(Future.succeededFuture(null)); 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<DeploymentConfig> dcCaptor = ArgumentCaptor.forClass(DeploymentConfig.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()); ArgumentCaptor<String> isNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> isNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<ImageStream> isCaptor = ArgumentCaptor.forClass(ImageStream.class); when(mockIsOps.reconcile(isNamespaceCaptor.capture(), isNameCaptor.capture(), isCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> bcNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> bcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<BuildConfig> bcCaptor = ArgumentCaptor.forClass(BuildConfig.class); when(mockBcOps.reconcile(bcNamespaceCaptor.capture(), bcNameCaptor.capture(), bcCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnectS2I()))); when(mockConnectS2IOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kcs2i)); when(mockConnectS2IOps.updateStatusAsync(any(KafkaConnectS2I.class))).thenReturn(Future.succeededFuture()); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockPdbOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockNetPolOps.reconcile(eq(kcs2i.getMetadata().getNamespace()), eq(KafkaConnectS2IResources.deploymentName(kcs2i.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); PlatformFeaturesAvailability pfa = new PlatformFeaturesAvailability(true, kubernetesVersion); KafkaConnectS2IAssemblyOperator ops = new KafkaConnectS2IAssemblyOperator(vertx, pfa, supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS)); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnectS2I.RESOURCE_KIND, kcs2iNamespace, kcs2iName), kcs2i) .onComplete(context.failing(v -> context.verify(() -> async.flag()))); }
@Test public void testUpdateClusterScaleUp(VertxTestContext context) { int scaleTo = 4; ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; CrdOperator mockConnectOps = supplier.connectOperator; DeploymentConfigOperator mockDcOps = supplier.deploymentConfigOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; BuildConfigOperator mockBcOps = supplier.buildConfigOperations; ImageStreamOperator mockIsOps = supplier.imagesStreamOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcs2iName = "foo"; String kcs2iNamespace = "test"; KafkaConnectS2I kcs2i = ResourceUtils.createEmptyKafkaConnectS2I(kcs2iNamespace, kcs2iName); KafkaConnectS2ICluster connect = KafkaConnectS2ICluster.fromCrd(kcs2i, VERSIONS); kcs2i.getSpec().setReplicas(scaleTo); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectS2IOps.get(kcs2iNamespace, kcs2iName)).thenReturn(kcs2i); when(mockConnectS2IOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kcs2i)); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(null)); when(mockConnectS2IOps.updateStatusAsync(any(KafkaConnectS2I.class))).thenReturn(Future.succeededFuture()); when(mockServiceOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateDeploymentConfig(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()); when(mockIsOps.get(kcs2iNamespace, KafkaConnectS2IResources.sourceImageStreamName(connect.getCluster()))).thenReturn(connect.generateSourceImageStream()); when(mockIsOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateTargetImageStream()); when(mockBcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateBuildConfig()); when(mockPdbOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generatePodDisruptionBudget()); when(mockServiceOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kcs2iNamespace, connect.getName(), scaleTo); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kcs2iNamespace, connect.getName(), scaleTo); when(mockConnectS2IOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnectS2I()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockIsOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockBcOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockPdbOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockNetPolOps.reconcile(eq(kcs2i.getMetadata().getNamespace()), eq(KafkaConnectS2IResources.deploymentName(kcs2i.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); KafkaConnectApi mockConnectClient = mock(KafkaConnectApi.class); when(mockConnectClient.list(anyString(), anyInt())).thenReturn(Future.succeededFuture(emptyList())); ConnectorPlugin plugin1 = new ConnectorPluginBuilder() .withConnectorClass("io.strimzi.MyClass") .withType("sink") .withVersion("1.0.0") .build(); when(mockConnectClient.listConnectorPlugins(anyString(), anyInt())).thenReturn(Future.succeededFuture(singletonList(plugin1))); when(mockConnectClient.updateConnectLoggers(anyString(), anyInt(), anyString(), any(OrderedProperties.class))).thenReturn(Future.succeededFuture()); PlatformFeaturesAvailability pfa = new PlatformFeaturesAvailability(true, kubernetesVersion); KafkaConnectS2IAssemblyOperator ops = new KafkaConnectS2IAssemblyOperator(vertx, pfa, supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnectS2I.RESOURCE_KIND, kcs2iNamespace, kcs2iName), kcs2i) .onComplete(context.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleUp(kcs2iNamespace, connect.getName(), scaleTo); async.flag(); }))); }
@Test public void testUpdateClusterScaleDown(VertxTestContext context) { int scaleTo = 2; ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; CrdOperator mockConnectOps = supplier.connectOperator; DeploymentConfigOperator mockDcOps = supplier.deploymentConfigOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; BuildConfigOperator mockBcOps = supplier.buildConfigOperations; ImageStreamOperator mockIsOps = supplier.imagesStreamOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcs2iName = "foo"; String kcs2iNamespace = "test"; KafkaConnectS2I kcs2i = ResourceUtils.createEmptyKafkaConnectS2I(kcs2iNamespace, kcs2iName); KafkaConnectS2ICluster connect = KafkaConnectS2ICluster.fromCrd(kcs2i, VERSIONS); kcs2i.getSpec().setReplicas(scaleTo); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectS2IOps.get(kcs2iNamespace, kcs2iName)).thenReturn(kcs2i); when(mockConnectS2IOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kcs2i)); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(null)); when(mockConnectS2IOps.updateStatusAsync(any(KafkaConnectS2I.class))).thenReturn(Future.succeededFuture()); when(mockServiceOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateDeploymentConfig(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()); when(mockIsOps.get(kcs2iNamespace, KafkaConnectS2IResources.sourceImageStreamName(connect.getCluster()))).thenReturn(connect.generateSourceImageStream()); when(mockIsOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateTargetImageStream()); when(mockBcOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generateBuildConfig()); when(mockPdbOps.get(kcs2iNamespace, connect.getName())).thenReturn(connect.generatePodDisruptionBudget()); when(mockServiceOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kcs2iNamespace, connect.getName(), 2); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kcs2iNamespace, connect.getName(), scaleTo); when(mockConnectS2IOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnectS2I()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockIsOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockBcOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockPdbOps.reconcile(any(), any(), any())).thenReturn(Future.succeededFuture()); when(mockNetPolOps.reconcile(eq(kcs2i.getMetadata().getNamespace()), eq(KafkaConnectS2IResources.deploymentName(kcs2i.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); KafkaConnectApi mockConnectClient = mock(KafkaConnectApi.class); when(mockConnectClient.list(anyString(), anyInt())).thenReturn(Future.succeededFuture(emptyList())); ConnectorPlugin plugin1 = new ConnectorPluginBuilder() .withConnectorClass("io.strimzi.MyClass") .withType("sink") .withVersion("1.0.0") .build(); when(mockConnectClient.listConnectorPlugins(anyString(), anyInt())).thenReturn(Future.succeededFuture(singletonList(plugin1))); when(mockConnectClient.updateConnectLoggers(anyString(), anyInt(), anyString(), any(OrderedProperties.class))).thenReturn(Future.succeededFuture()); PlatformFeaturesAvailability pfa = new PlatformFeaturesAvailability(true, kubernetesVersion); KafkaConnectS2IAssemblyOperator ops = new KafkaConnectS2IAssemblyOperator(vertx, pfa, supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnectS2I.RESOURCE_KIND, kcs2iNamespace, kcs2iName), kcs2i) .onComplete(context.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleDown(kcs2iNamespace, connect.getName(), scaleTo); async.flag(); }))); }