KafkaConnectS2ICluster extends KafkaConnectCluster { public static KafkaConnectS2ICluster fromCrd(KafkaConnectS2I kafkaConnectS2I, KafkaVersion.Lookup versions) { KafkaConnectS2ISpec spec = kafkaConnectS2I.getSpec(); KafkaConnectS2ICluster cluster = fromSpec(spec, versions, new KafkaConnectS2ICluster(kafkaConnectS2I)); cluster.setOwnerReference(kafkaConnectS2I); cluster.setInsecureSourceRepository(spec.isInsecureSourceRepository()); cluster.setBuildResourceRequirements(spec.getBuildResources()); return cluster; } private  KafkaConnectS2ICluster(HasMetadata resource); static KafkaConnectS2ICluster fromCrd(KafkaConnectS2I kafkaConnectS2I, KafkaVersion.Lookup versions); DeploymentConfig generateDeploymentConfig(Map<String, String> annotations, boolean isOpenShift, ImagePullPolicy imagePullPolicy,
                                                     List<LocalObjectReference> imagePullSecrets); ImageStream generateSourceImageStream(); ImageStream generateTargetImageStream(); BuildConfig generateBuildConfig(); boolean isInsecureSourceRepository(); void setInsecureSourceRepository(boolean insecureSourceRepository); void setBuildResourceRequirements(ResourceRequirements buildRequirements);  }
@Test public void testFromCrd() { assertThat(kc.image, is(KafkaConnectS2IResources.deploymentName(cluster) + ":latest")); assertThat(kc.replicas, is(replicas)); assertThat(kc.sourceImageBaseName + ":" + kc.sourceImageTag, is(image)); assertThat(kc.readinessProbeOptions.getInitialDelaySeconds(), is(healthDelay)); assertThat(kc.readinessProbeOptions.getTimeoutSeconds(), is(healthTimeout)); assertThat(kc.livenessProbeOptions.getInitialDelaySeconds(), is(healthDelay)); assertThat(kc.livenessProbeOptions.getTimeoutSeconds(), is(healthTimeout)); assertThat(kc.getConfiguration().asOrderedProperties(), is(expectedConfiguration)); assertThat(kc.bootstrapServers, is(bootstrapServers)); assertThat(kc.isInsecureSourceRepository(), is(false)); }
@Test public void testGenerateServiceWithoutMetrics() { KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .editSpec() .withMetrics(null) .endSpec() .build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); Service svc = kc.generateService(); assertThat(svc.getSpec().getType(), is("ClusterIP")); assertThat(svc.getMetadata().getLabels(), is(expectedLabels(kc.getServiceName()))); assertThat(svc.getSpec().getSelector(), is(expectedSelectorLabels())); assertThat(svc.getSpec().getPorts().size(), is(1)); assertThat(svc.getSpec().getPorts().get(0).getPort(), is(Integer.valueOf(KafkaConnectCluster.REST_API_PORT))); assertThat(svc.getSpec().getPorts().get(0).getName(), is(KafkaConnectCluster.REST_API_PORT_NAME)); assertThat(svc.getSpec().getPorts().get(0).getProtocol(), is("TCP")); assertThat(svc.getMetadata().getAnnotations().containsKey("prometheus.io/port"), is(false)); assertThat(svc.getMetadata().getAnnotations().containsKey("prometheus.io/scrape"), is(false)); assertThat(svc.getMetadata().getAnnotations().containsKey("prometheus.io/path"), is(false)); checkOwnerReference(kc.createOwnerReference(), svc); }
@Test public void testImagePullSecretsCO() { LocalObjectReference secret1 = new LocalObjectReference("some-pull-secret"); LocalObjectReference secret2 = new LocalObjectReference("some-other-pull-secret"); List<LocalObjectReference> secrets = new ArrayList<>(2); secrets.add(secret1); secrets.add(secret2); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(this.resource, VERSIONS); Deployment dep = kc.generateDeployment(emptyMap(), true, null, secrets); assertThat(dep.getSpec().getTemplate().getSpec().getImagePullSecrets().size(), is(2)); assertThat(dep.getSpec().getTemplate().getSpec().getImagePullSecrets().contains(secret1), is(true)); assertThat(dep.getSpec().getTemplate().getSpec().getImagePullSecrets().contains(secret2), is(true)); }
@Test public void testImagePullSecretsBoth() { LocalObjectReference secret1 = new LocalObjectReference("some-pull-secret"); LocalObjectReference secret2 = new LocalObjectReference("some-other-pull-secret"); KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .editSpec() .withNewTemplate() .withNewPod() .withImagePullSecrets(secret2) .endPod() .endTemplate() .endSpec() .build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); Deployment dep = kc.generateDeployment(emptyMap(), true, null, singletonList(secret1)); assertThat(dep.getSpec().getTemplate().getSpec().getImagePullSecrets().size(), is(1)); assertThat(dep.getSpec().getTemplate().getSpec().getImagePullSecrets().contains(secret1), is(false)); assertThat(dep.getSpec().getTemplate().getSpec().getImagePullSecrets().contains(secret2), is(true)); }
@Test public void testPodDisruptionBudget() { KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .editSpec() .withNewTemplate() .withNewPodDisruptionBudget() .withMaxUnavailable(2) .endPodDisruptionBudget() .endTemplate() .endSpec() .build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); PodDisruptionBudget pdb = kc.generatePodDisruptionBudget(); assertThat(pdb.getSpec().getMaxUnavailable(), is(new IntOrString(2))); }
@Test public void testDefaultPodDisruptionBudget() { KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource).build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); PodDisruptionBudget pdb = kc.generatePodDisruptionBudget(); assertThat(pdb.getSpec().getMaxUnavailable(), is(new IntOrString(1))); }
@Test public void testKafkaConnectContainerEnvVars() { ContainerEnvVar envVar1 = new ContainerEnvVar(); String testEnvOneKey = "TEST_ENV_1"; String testEnvOneValue = "test.env.one"; envVar1.setName(testEnvOneKey); envVar1.setValue(testEnvOneValue); ContainerEnvVar envVar2 = new ContainerEnvVar(); String testEnvTwoKey = "TEST_ENV_2"; String testEnvTwoValue = "test.env.two"; envVar2.setName(testEnvTwoKey); envVar2.setValue(testEnvTwoValue); List<ContainerEnvVar> testEnvs = new ArrayList<>(); testEnvs.add(envVar1); testEnvs.add(envVar2); ContainerTemplate kafkaConnectContainer = new ContainerTemplate(); kafkaConnectContainer.setEnv(testEnvs); KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .editSpec() .withNewTemplate() .withConnectContainer(kafkaConnectContainer) .endTemplate() .endSpec() .build(); List<EnvVar> kafkaEnvVars = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS).getEnvVars(); assertThat("Failed to correctly set container environment variable: " + testEnvOneKey, kafkaEnvVars.stream().filter(env -> testEnvOneKey.equals(env.getName())) .map(EnvVar::getValue).findFirst().orElse("").equals(testEnvOneValue), is(true)); assertThat("Failed to correctly set container environment variable: " + testEnvTwoKey, kafkaEnvVars.stream().filter(env -> testEnvTwoKey.equals(env.getName())) .map(EnvVar::getValue).findFirst().orElse("").equals(testEnvTwoValue), is(true)); }
@Test public void testKafkaContainerEnvVarsConflict() { ContainerEnvVar envVar1 = new ContainerEnvVar(); String testEnvOneKey = KafkaConnectCluster.ENV_VAR_KAFKA_CONNECT_CONFIGURATION; String testEnvOneValue = "test.env.one"; envVar1.setName(testEnvOneKey); envVar1.setValue(testEnvOneValue); ContainerEnvVar envVar2 = new ContainerEnvVar(); String testEnvTwoKey = KafkaConnectCluster.ENV_VAR_KAFKA_CONNECT_BOOTSTRAP_SERVERS; String testEnvTwoValue = "test.env.two"; envVar2.setName(testEnvTwoKey); envVar2.setValue(testEnvTwoValue); List<ContainerEnvVar> testEnvs = new ArrayList<>(); testEnvs.add(envVar1); testEnvs.add(envVar2); ContainerTemplate kafkaConnectContainer = new ContainerTemplate(); kafkaConnectContainer.setEnv(testEnvs); KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .editSpec() .withNewTemplate() .withConnectContainer(kafkaConnectContainer) .endTemplate() .endSpec() .build(); List<EnvVar> kafkaEnvVars = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS).getEnvVars(); assertThat("Failed to prevent over writing existing container environment variable: " + testEnvOneKey, kafkaEnvVars.stream().filter(env -> testEnvOneKey.equals(env.getName())) .map(EnvVar::getValue).findFirst().orElse("").equals(testEnvOneValue), is(false)); assertThat("Failed to prevent over writing existing container environment variable: " + testEnvTwoKey, kafkaEnvVars.stream().filter(env -> testEnvTwoKey.equals(env.getName())) .map(EnvVar::getValue).findFirst().orElse("").equals(testEnvTwoValue), is(false)); }
@Test public void testGenerateDeploymentWithOAuthWithMissingClientSecret() { assertThrows(InvalidResourceException.class, () -> { KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .editSpec() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .build()) .endSpec() .build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithOAuthWithMissingUri() { assertThrows(InvalidResourceException.class, () -> { KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .editSpec() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withNewClientSecret() .withSecretName("my-secret-secret") .withKey("my-secret-key") .endClientSecret() .build()) .endSpec() .build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); }); }
@Test public void testNetworkPolicyWithConnectorOperator() { KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); NetworkPolicy np = kc.generateNetworkPolicy(true, true); assertThat(np.getMetadata().getName(), is(kc.getName())); assertThat(np.getSpec().getPodSelector().getMatchLabels(), is(kc.getSelectorLabels().toMap())); assertThat(np.getSpec().getIngress().size(), is(2)); assertThat(np.getSpec().getIngress().get(0).getPorts().size(), is(1)); assertThat(np.getSpec().getIngress().get(0).getPorts().get(0).getPort().getIntVal(), is(KafkaConnectCluster.REST_API_PORT)); assertThat(np.getSpec().getIngress().get(0).getFrom().size(), is(2)); assertThat(np.getSpec().getIngress().get(0).getFrom().get(0).getPodSelector().getMatchLabels(), is(kc.getSelectorLabels().toMap())); assertThat(np.getSpec().getIngress().get(0).getFrom().get(0).getNamespaceSelector(), is(nullValue())); assertThat(np.getSpec().getIngress().get(0).getFrom().get(1).getPodSelector().getMatchLabels(), is(singletonMap(Labels.STRIMZI_KIND_LABEL, "cluster-operator"))); assertThat(np.getSpec().getIngress().get(0).getFrom().get(1).getNamespaceSelector().getMatchLabels(), is(nullValue())); assertThat(np.getSpec().getIngress().get(1).getPorts().size(), is(1)); assertThat(np.getSpec().getIngress().get(1).getPorts().get(0).getPort().getIntVal(), is(KafkaConnectCluster.METRICS_PORT)); }
@Test public void testNetworkPolicyWithoutConnectorOperator() { KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource) .build(); KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS); assertThat(kc.generateNetworkPolicy(true, false), is(nullValue())); }