KafkaConnectAssemblyOperator extends AbstractConnectOperator<KubernetesClient, KafkaConnect, KafkaConnectList, DoneableKafkaConnect, Resource<KafkaConnect, DoneableKafkaConnect>, KafkaConnectSpec, KafkaConnectStatus> { @Override protected Future<KafkaConnectStatus> createOrUpdate(Reconciliation reconciliation, KafkaConnect kafkaConnect) { KafkaConnectCluster connect; KafkaConnectStatus kafkaConnectStatus = new KafkaConnectStatus(); try { connect = KafkaConnectCluster.fromCrd(kafkaConnect, versions); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(kafkaConnect, kafkaConnectStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(kafkaConnectStatus, e)); } Promise<KafkaConnectStatus> createOrUpdatePromise = Promise.promise(); String namespace = reconciliation.namespace(); 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); log.debug("{}: Updating Kafka Connect cluster", reconciliation); Future<KafkaConnectS2I> connectS2ICheck; if (connectS2IOperations != null) { connectS2ICheck = connectS2IOperations.getAsync(kafkaConnect.getMetadata().getNamespace(), kafkaConnect.getMetadata().getName()); } else { connectS2ICheck = Future.succeededFuture(null); } boolean connectHasZeroReplicas = connect.getReplicas() == 0; connectS2ICheck .compose(otherConnect -> { if (otherConnect != null && kafkaConnect.getMetadata().getCreationTimestamp().compareTo(otherConnect.getMetadata().getCreationTimestamp()) > 0) { return Future.failedFuture("Both KafkaConnect and KafkaConnectS2I exist with the same name. " + "KafkaConnectS2I is older and will be used while this custom resource will be ignored."); } else { return Future.succeededFuture(); } }) .compose(i -> connectServiceAccount(namespace, connect)) .compose(i -> connectInitServiceAccount(namespace, connect)) .compose(i -> connectInitClusterRoleBinding(namespace, kafkaConnect.getMetadata().getName(), connect)) .compose(i -> networkPolicyOperator.reconcile(namespace, connect.getName(), connect.generateNetworkPolicy(pfa.isNamespaceAndPodSelectorNetworkPolicySupported(), isUseResources(kafkaConnect)))) .compose(i -> deploymentOperations.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 -> podDisruptionBudgetOperator.reconcile(namespace, connect.getName(), connect.generatePodDisruptionBudget())) .compose(i -> deploymentOperations.reconcile(namespace, connect.getName(), connect.generateDeployment(annotations, pfa.isOpenshift(), imagePullPolicy, imagePullSecrets))) .compose(i -> deploymentOperations.scaleUp(namespace, connect.getName(), connect.getReplicas())) .compose(i -> deploymentOperations.waitForObserved(namespace, connect.getName(), 1_000, operationTimeoutMs)) .compose(i -> connectHasZeroReplicas ? Future.succeededFuture() : deploymentOperations.readiness(namespace, connect.getName(), 1_000, operationTimeoutMs)) .compose(i -> reconcileConnectors(reconciliation, kafkaConnect, kafkaConnectStatus, connectHasZeroReplicas, desiredLogging, connect.getDefaultLogConfig())) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(kafkaConnect, kafkaConnectStatus, reconciliationResult); if (!connectHasZeroReplicas) { kafkaConnectStatus.setUrl(KafkaConnectResources.url(connect.getCluster(), namespace, KafkaConnectCluster.REST_API_PORT)); } kafkaConnectStatus.setReplicas(connect.getReplicas()); kafkaConnectStatus.setLabelSelector(connect.getSelectorLabels().toSelectorString()); if (reconciliationResult.succeeded()) { createOrUpdatePromise.complete(kafkaConnectStatus); } else { createOrUpdatePromise.fail(new ReconciliationException(kafkaConnectStatus, reconciliationResult.cause())); } }); return createOrUpdatePromise.future(); } KafkaConnectAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                        ResourceOperatorSupplier supplier,
                                        ClusterOperatorConfig config);  KafkaConnectAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                        ResourceOperatorSupplier supplier,
                                        ClusterOperatorConfig config,
                                        Function<Vertx, KafkaConnectApi> connectClientProvider);  KafkaConnectAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
                                        ResourceOperatorSupplier supplier,
                                        ClusterOperatorConfig config,
                                        Function<Vertx, KafkaConnectApi> connectClientProvider, int port);   }
@Test public void testCreateOrUpdateDoesNotUpdateWithNoDiff(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectOps = supplier.connectOperator; CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcName = "foo"; String kcNamespace = "test"; KafkaConnect kc = ResourceUtils.createEmptyKafkaConnect(kcNamespace, kcName); KafkaConnectCluster connect = KafkaConnectCluster.fromCrd(kc, VERSIONS); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectOps.get(kcNamespace, kcName)).thenReturn(kc); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kc)); when(mockConnectOps.updateStatusAsync(any(KafkaConnect.class))).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.getAsync(kcNamespace, kcName)).thenReturn(Future.succeededFuture(null)); when(mockServiceOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> serviceNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Service> serviceCaptor = ArgumentCaptor.forClass(Service.class); when(mockServiceOps.reconcile(eq(kcNamespace), serviceNameCaptor.capture(), serviceCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(eq(kcNamespace), 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(kcNamespace), 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(kcNamespace), dcScaleDownNameCaptor.capture(), dcScaleDownReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockNetPolOps.reconcile(eq(kc.getMetadata().getNamespace()), eq(KafkaConnectResources.deploymentName(kc.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); ArgumentCaptor<PodDisruptionBudget> pdbCaptor = ArgumentCaptor.forClass(PodDisruptionBudget.class); when(mockPdbOps.reconcile(anyString(), any(), pdbCaptor.capture())).thenReturn(Future.succeededFuture()); 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()); KafkaConnectAssemblyOperator ops = new KafkaConnectAssemblyOperator(vertx, new PlatformFeaturesAvailability(true, kubernetesVersion), supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnect.RESOURCE_KIND, kcNamespace, kcName), kc) .onComplete(context.succeeding(v -> context.verify(() -> { List<Service> capturedServices = serviceCaptor.getAllValues(); assertThat(capturedServices, hasSize(1)); List<Deployment> capturedDc = dcCaptor.getAllValues(); assertThat(capturedDc, hasSize(1)); assertThat(dcScaleDownNameCaptor.getAllValues(), hasSize(1)); assertThat(dcScaleUpNameCaptor.getAllValues(), hasSize(1)); List<PodDisruptionBudget> capturedPdb = pdbCaptor.getAllValues(); assertThat(capturedPdb, hasSize(1)); PodDisruptionBudget pdb = capturedPdb.get(0); assertThat(pdb.getMetadata().getName(), is(connect.getName())); assertThat(pdb, is(connect.generatePodDisruptionBudget())); async.flag(); }))); }
@Test public void testCreateOrUpdateUpdatesCluster(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectOps = supplier.connectOperator; CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcName = "foo"; String kcNamespace = "test"; KafkaConnect kc = ResourceUtils.createEmptyKafkaConnect(kcNamespace, kcName); KafkaConnectCluster connect = KafkaConnectCluster.fromCrd(kc, VERSIONS); kc.getSpec().setImage("some/different:image"); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectOps.get(kcNamespace, kcName)).thenReturn(kc); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kc)); when(mockConnectOps.updateStatusAsync(any(KafkaConnect.class))).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.getAsync(kcNamespace, kcName)).thenReturn(Future.succeededFuture(null)); when(mockServiceOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> serviceNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Service> serviceCaptor = ArgumentCaptor.forClass(Service.class); when(mockServiceOps.reconcile(eq(kcNamespace), serviceNameCaptor.capture(), serviceCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(eq(kcNamespace), 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(kcNamespace), 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(kcNamespace), dcScaleDownNameCaptor.capture(), dcScaleDownReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockNetPolOps.reconcile(eq(kc.getMetadata().getNamespace()), eq(KafkaConnectResources.deploymentName(kc.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); ArgumentCaptor<PodDisruptionBudget> pdbCaptor = ArgumentCaptor.forClass(PodDisruptionBudget.class); when(mockPdbOps.reconcile(anyString(), any(), pdbCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockConnectOps.get(kcNamespace, kcName)).thenReturn(kc); ConfigMap metricsCm = new ConfigMapBuilder().withNewMetadata() .withName(KafkaConnectResources.metricsAndLogConfigMapName(kcName)) .withNamespace(kcNamespace) .endMetadata() .withData(Collections.singletonMap(AbstractModel.ANCILLARY_CM_KEY_METRICS, METRICS_CONFIG)) .build(); when(mockCmOps.get(kcNamespace, KafkaConnectResources.metricsAndLogConfigMapName(kcName))).thenReturn(metricsCm); ConfigMap loggingCm = new ConfigMapBuilder().withNewMetadata() .withName(KafkaConnectResources.metricsAndLogConfigMapName(kcName)) .withNamespace(kcNamespace) .endMetadata() .withData(Collections.singletonMap(AbstractModel.ANCILLARY_CM_KEY_LOG_CONFIG, LOGGING_CONFIG)) .build(); when(mockCmOps.get(kcNamespace, KafkaConnectResources.metricsAndLogConfigMapName(kcName))).thenReturn(metricsCm); Set<String> metricsCms = TestUtils.set(); doAnswer(invocation -> { metricsCms.add(invocation.getArgument(1)); return Future.succeededFuture(); }).when(mockCmOps).reconcile(eq(kcNamespace), 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()); KafkaConnectAssemblyOperator ops = new KafkaConnectAssemblyOperator(vertx, new PlatformFeaturesAvailability(true, kubernetesVersion), supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnect.RESOURCE_KIND, kcNamespace, kcName), kc) .onComplete(context.succeeding(v -> context.verify(() -> { KafkaConnectCluster compareTo = KafkaConnectCluster.fromCrd(kc, VERSIONS); List<Service> capturedServices = serviceCaptor.getAllValues(); assertThat(capturedServices, hasSize(1)); Service service = capturedServices.get(0); assertThat(service.getMetadata().getName(), is(compareTo.getServiceName())); assertThat(service, is(compareTo.generateService())); List<Deployment> capturedDc = dcCaptor.getAllValues(); assertThat(capturedDc, hasSize(1)); Deployment dc = capturedDc.get(0); assertThat(dc.getMetadata().getName(), is(compareTo.getName())); Map<String, String> annotations = new HashMap(); annotations.put(Annotations.ANNO_STRIMZI_LOGGING_DYNAMICALLY_UNCHANGEABLE_HASH, Util.stringHash(Util.getLoggingDynamicallyUnmodifiableEntries(loggingCm.getData().get(compareTo.ANCILLARY_CM_KEY_LOG_CONFIG)))); assertThat(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(connect.getName())); assertThat(pdb, is(connect.generatePodDisruptionBudget())); assertThat(dcScaleDownNameCaptor.getAllValues(), hasSize(1)); assertThat(dcScaleUpNameCaptor.getAllValues(), hasSize(1)); verify(mockCmOps, never()).createOrUpdate(any()); async.flag(); }))); }
@Test public void testCreateOrUpdateFailsWhenDeploymentUpdateFails(VertxTestContext context) { ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectOps = supplier.connectOperator; CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; String kcName = "foo"; String kcNamespace = "test"; KafkaConnect kc = ResourceUtils.createEmptyKafkaConnect(kcNamespace, kcName); KafkaConnectCluster connect = KafkaConnectCluster.fromCrd(kc, VERSIONS); kc.getSpec().setImage("some/different:image"); when(mockConnectOps.get(kcNamespace, kcName)).thenReturn(kc); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kc)); when(mockConnectOps.updateStatusAsync(any(KafkaConnect.class))).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.getAsync(kcNamespace, kcName)).thenReturn(Future.succeededFuture(null)); when(mockServiceOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> serviceNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> serviceNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Service> serviceCaptor = ArgumentCaptor.forClass(Service.class); when(mockServiceOps.reconcile(serviceNamespaceCaptor.capture(), serviceNameCaptor.capture(), serviceCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> dcNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Deployment> dcCaptor = ArgumentCaptor.forClass(Deployment.class); when(mockDcOps.reconcile(dcNamespaceCaptor.capture(), dcNameCaptor.capture(), dcCaptor.capture())).thenReturn(Future.failedFuture("Failed")); ArgumentCaptor<String> dcScaleUpNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> dcScaleUpNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleUpReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleUp(dcScaleUpNamespaceCaptor.capture(), dcScaleUpNameCaptor.capture(), dcScaleUpReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> dcScaleDownNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> dcScaleDownNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Integer> dcScaleDownReplicasCaptor = ArgumentCaptor.forClass(Integer.class); when(mockDcOps.scaleDown(dcScaleDownNamespaceCaptor.capture(), dcScaleDownNameCaptor.capture(), dcScaleDownReplicasCaptor.capture())).thenReturn(Future.succeededFuture()); when(mockNetPolOps.reconcile(eq(kc.getMetadata().getNamespace()), eq(KafkaConnectResources.deploymentName(kc.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); when(mockConnectOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnect()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new PodDisruptionBudget()))); KafkaConnectAssemblyOperator ops = new KafkaConnectAssemblyOperator(vertx, new PlatformFeaturesAvailability(true, kubernetesVersion), supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS)); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnect.RESOURCE_KIND, kcNamespace, kcName), kc) .onComplete(context.failing(v -> async.flag())); }
@Test public void testUpdateClusterScaleUp(VertxTestContext context) { final int scaleTo = 4; ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectOps = supplier.connectOperator; CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcName = "foo"; String kcNamespace = "test"; KafkaConnect kc = ResourceUtils.createEmptyKafkaConnect(kcNamespace, kcName); KafkaConnectCluster connect = KafkaConnectCluster.fromCrd(kc, VERSIONS); kc.getSpec().setReplicas(scaleTo); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectOps.get(kcNamespace, kcName)).thenReturn(kc); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kc)); when(mockConnectOps.updateStatusAsync(any(KafkaConnect.class))).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.getAsync(kcNamespace, kcName)).thenReturn(Future.succeededFuture(null)); when(mockServiceOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockServiceOps.reconcile(eq(kcNamespace), any(), any())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(eq(kcNamespace), any(), any())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kcNamespace, connect.getName(), scaleTo); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kcNamespace, connect.getName(), scaleTo); when(mockNetPolOps.reconcile(eq(kc.getMetadata().getNamespace()), eq(KafkaConnectResources.deploymentName(kc.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); when(mockConnectOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnect()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new PodDisruptionBudget()))); 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()); KafkaConnectAssemblyOperator ops = new KafkaConnectAssemblyOperator(vertx, new PlatformFeaturesAvailability(true, kubernetesVersion), supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnect.RESOURCE_KIND, kcNamespace, kcName), kc) .onComplete(context.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleUp(kcNamespace, connect.getName(), scaleTo); async.flag(); }))); }
@Test public void testUpdateClusterScaleDown(VertxTestContext context) { int scaleTo = 2; ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(true); CrdOperator mockConnectOps = supplier.connectOperator; CrdOperator mockConnectS2IOps = supplier.connectS2IOperator; DeploymentOperator mockDcOps = supplier.deploymentOperations; PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator; ConfigMapOperator mockCmOps = supplier.configMapOperations; ServiceOperator mockServiceOps = supplier.serviceOperations; NetworkPolicyOperator mockNetPolOps = supplier.networkPolicyOperator; CrdOperator<KubernetesClient, KafkaConnector, KafkaConnectorList, DoneableKafkaConnector> mockConnectorOps = supplier.kafkaConnectorOperator; String kcName = "foo"; String kcNamespace = "test"; KafkaConnect kc = ResourceUtils.createEmptyKafkaConnect(kcNamespace, kcName); KafkaConnectCluster connect = KafkaConnectCluster.fromCrd(kc, VERSIONS); kc.getSpec().setReplicas(scaleTo); when(mockConnectorOps.listAsync(anyString(), any(Optional.class))).thenReturn(Future.succeededFuture(emptyList())); when(mockConnectOps.get(kcNamespace, kcName)).thenReturn(kc); when(mockConnectOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(kc)); when(mockConnectOps.updateStatusAsync(any(KafkaConnect.class))).thenReturn(Future.succeededFuture()); when(mockConnectS2IOps.getAsync(kcNamespace, kcName)).thenReturn(Future.succeededFuture(null)); when(mockServiceOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateService()); when(mockDcOps.get(kcNamespace, connect.getName())).thenReturn(connect.generateDeployment(new HashMap<String, String>(), true, null, null)); when(mockDcOps.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockDcOps.waitForObserved(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(mockServiceOps.reconcile(eq(kcNamespace), any(), any())).thenReturn(Future.succeededFuture()); when(mockDcOps.reconcile(eq(kcNamespace), any(), any())).thenReturn(Future.succeededFuture()); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleUp(kcNamespace, connect.getName(), scaleTo); doAnswer(i -> Future.succeededFuture(scaleTo)) .when(mockDcOps).scaleDown(kcNamespace, connect.getName(), scaleTo); when(mockNetPolOps.reconcile(eq(kc.getMetadata().getNamespace()), eq(KafkaConnectResources.deploymentName(kc.getMetadata().getName())), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new NetworkPolicy()))); when(mockConnectOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new KafkaConnect()))); when(mockCmOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new ConfigMap()))); when(mockPdbOps.reconcile(anyString(), any(), any())).thenReturn(Future.succeededFuture(ReconcileResult.created(new PodDisruptionBudget()))); 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()); KafkaConnectAssemblyOperator ops = new KafkaConnectAssemblyOperator(vertx, new PlatformFeaturesAvailability(true, kubernetesVersion), supplier, ResourceUtils.dummyClusterOperatorConfig(VERSIONS), x -> mockConnectClient); Checkpoint async = context.checkpoint(); ops.createOrUpdate(new Reconciliation("test-trigger", KafkaConnect.RESOURCE_KIND, kcNamespace, kcName), kc) .onComplete(context.succeeding(v -> context.verify(() -> { verify(mockDcOps).scaleUp(kcNamespace, connect.getName(), scaleTo); async.flag(); }))); }