@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(); }))); }
@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(); }
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(); } }
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); }
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); }
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);  }