KafkaMirrorMaker2Cluster extends KafkaConnectCluster { public static KafkaMirrorMaker2Cluster fromCrd(KafkaMirrorMaker2 kafkaMirrorMaker2, KafkaVersion.Lookup versions) { KafkaMirrorMaker2Cluster cluster = new KafkaMirrorMaker2Cluster(kafkaMirrorMaker2); KafkaMirrorMaker2Spec spec = kafkaMirrorMaker2.getSpec(); cluster.setOwnerReference(kafkaMirrorMaker2); String specVersion = spec.getVersion(); if (versions.version(specVersion).compareVersion("2.4.0") < 0) { throw new InvalidResourceException("Kafka MirrorMaker 2.0 is not available in the version at spec.version (" + specVersion + "). Kafka MirrorMaker 2.0 is available in Kafka version 2.4.0 and later."); } cluster.setImage(versions.kafkaMirrorMaker2Version(spec.getImage(), specVersion)); List<KafkaMirrorMaker2ClusterSpec> clustersList = ModelUtils.asListOrEmptyList(spec.getClusters()); cluster.setClusters(clustersList); KafkaMirrorMaker2ClusterSpec connectCluster = new KafkaMirrorMaker2ClusterSpecBuilder().build(); String connectClusterAlias = spec.getConnectCluster(); if (connectClusterAlias != null) { connectCluster = clustersList.stream() .filter(clustersListItem -> spec.getConnectCluster().equals(clustersListItem.getAlias())) .findFirst() .orElseThrow(() -> new InvalidResourceException("connectCluster with alias " + connectClusterAlias + " cannot be found in the list of clusters at spec.clusters")); } cluster.setConfiguration(new KafkaMirrorMaker2Configuration(connectCluster.getConfig().entrySet())); return fromSpec(buildKafkaConnectSpec(spec, connectCluster), versions, cluster); } private  KafkaMirrorMaker2Cluster(HasMetadata resource); static KafkaMirrorMaker2Cluster fromCrd(KafkaMirrorMaker2 kafkaMirrorMaker2, 
                                                   KafkaVersion.Lookup versions); @Override String getServiceAccountName();  }
@Test public void testDefaultValues() { KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(ResourceUtils.createEmptyKafkaMirrorMaker2(namespace, cluster), VERSIONS); assertThat(kmm2.image, is(KafkaVersionTestUtils.DEFAULT_KAFKA_CONNECT_IMAGE)); assertThat(kmm2.replicas, is(KafkaMirrorMaker2Cluster.DEFAULT_REPLICAS)); assertThat(kmm2.readinessProbeOptions.getInitialDelaySeconds(), is(KafkaMirrorMaker2Cluster.DEFAULT_HEALTHCHECK_DELAY)); assertThat(kmm2.readinessProbeOptions.getTimeoutSeconds(), is(KafkaMirrorMaker2Cluster.DEFAULT_HEALTHCHECK_TIMEOUT)); assertThat(kmm2.livenessProbeOptions.getInitialDelaySeconds(), is(KafkaMirrorMaker2Cluster.DEFAULT_HEALTHCHECK_DELAY)); assertThat(kmm2.livenessProbeOptions.getTimeoutSeconds(), is(KafkaMirrorMaker2Cluster.DEFAULT_HEALTHCHECK_TIMEOUT)); assertThat(kmm2.getConfiguration().asOrderedProperties(), is(defaultConfiguration)); }
@Test public void testFromCrd() { assertThat(kmm2.replicas, is(replicas)); assertThat(kmm2.image, is(image)); assertThat(kmm2.readinessProbeOptions.getInitialDelaySeconds(), is(healthDelay)); assertThat(kmm2.readinessProbeOptions.getTimeoutSeconds(), is(healthTimeout)); assertThat(kmm2.livenessProbeOptions.getInitialDelaySeconds(), is(healthDelay)); assertThat(kmm2.livenessProbeOptions.getTimeoutSeconds(), is(healthTimeout)); assertThat(kmm2.getConfiguration().asOrderedProperties(), is(expectedConfiguration)); assertThat(kmm2.bootstrapServers, is(bootstrapServers)); }
@Test public void testGenerateServiceWithoutMetrics() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withMetrics(null) .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Service svc = kmm2.generateService(); assertThat(svc.getSpec().getType(), is("ClusterIP")); assertThat(svc.getMetadata().getLabels(), is(expectedLabels(kmm2.getServiceName()))); assertThat(svc.getSpec().getSelector(), is(expectedSelectorLabels())); assertThat(svc.getSpec().getPorts().size(), is(1)); assertThat(svc.getSpec().getPorts().get(0).getPort(), is(Integer.valueOf(KafkaMirrorMaker2Cluster.REST_API_PORT))); assertThat(svc.getSpec().getPorts().get(0).getName(), is(KafkaMirrorMaker2Cluster.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(kmm2.createOwnerReference(), svc); }
@Test public void testTemplate() { Map<String, String> depLabels = TestUtils.map("l1", "v1", "l2", "v2", Labels.KUBERNETES_PART_OF_LABEL, "custom-part", Labels.KUBERNETES_MANAGED_BY_LABEL, "custom-managed-by"); Map<String, String> expectedDepLabels = new HashMap<>(depLabels); expectedDepLabels.remove(Labels.KUBERNETES_MANAGED_BY_LABEL); Map<String, String> depAnots = TestUtils.map("a1", "v1", "a2", "v2"); Map<String, String> podLabels = TestUtils.map("l3", "v3", "l4", "v4"); Map<String, String> podAnots = TestUtils.map("a3", "v3", "a4", "v4"); Map<String, String> svcLabels = TestUtils.map("l5", "v5", "l6", "v6"); Map<String, String> svcAnots = TestUtils.map("a5", "v5", "a6", "v6"); Map<String, String> pdbLabels = TestUtils.map("l7", "v7", "l8", "v8"); Map<String, String> pdbAnots = TestUtils.map("a7", "v7", "a8", "v8"); HostAlias hostAlias1 = new HostAliasBuilder() .withHostnames("my-host-1", "my-host-2") .withIp("192.168.1.86") .build(); HostAlias hostAlias2 = new HostAliasBuilder() .withHostnames("my-host-3") .withIp("192.168.1.87") .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewTemplate() .withNewDeployment() .withNewMetadata() .withLabels(depLabels) .withAnnotations(depAnots) .endMetadata() .endDeployment() .withNewPod() .withNewMetadata() .withLabels(podLabels) .withAnnotations(podAnots) .endMetadata() .withNewPriorityClassName("top-priority") .withNewSchedulerName("my-scheduler") .withHostAliases(hostAlias1, hostAlias2) .endPod() .withNewApiService() .withNewMetadata() .withLabels(svcLabels) .withAnnotations(svcAnots) .endMetadata() .endApiService() .withNewPodDisruptionBudget() .withNewMetadata() .withLabels(pdbLabels) .withAnnotations(pdbAnots) .endMetadata() .endPodDisruptionBudget() .endTemplate() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); assertThat(dep.getMetadata().getLabels().entrySet().containsAll(expectedDepLabels.entrySet()), is(true)); assertThat(dep.getMetadata().getAnnotations().entrySet().containsAll(depAnots.entrySet()), is(true)); assertThat(dep.getSpec().getTemplate().getSpec().getPriorityClassName(), is("top-priority")); assertThat(dep.getSpec().getTemplate().getMetadata().getLabels().entrySet().containsAll(podLabels.entrySet()), is(true)); assertThat(dep.getSpec().getTemplate().getMetadata().getAnnotations().entrySet().containsAll(podAnots.entrySet()), is(true)); assertThat(dep.getSpec().getTemplate().getSpec().getSchedulerName(), is("my-scheduler")); assertThat(dep.getSpec().getTemplate().getSpec().getHostAliases(), containsInAnyOrder(hostAlias1, hostAlias2)); Service svc = kmm2.generateService(); assertThat(svc.getMetadata().getLabels().entrySet().containsAll(svcLabels.entrySet()), is(true)); assertThat(svc.getMetadata().getAnnotations().entrySet().containsAll(svcAnots.entrySet()), is(true)); PodDisruptionBudget pdb = kmm2.generatePodDisruptionBudget(); assertThat(pdb.getMetadata().getLabels().entrySet().containsAll(pdbLabels.entrySet()), is(true)); assertThat(pdb.getMetadata().getAnnotations().entrySet().containsAll(pdbAnots.entrySet()), is(true)); }
@Test public void testExternalConfigurationSecretEnvs() { ExternalConfigurationEnv env = new ExternalConfigurationEnvBuilder() .withName("MY_ENV_VAR") .withNewValueFrom() .withSecretKeyRef(new SecretKeySelectorBuilder().withName("my-secret").withKey("my-key").withOptional(false).build()) .endValueFrom() .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewExternalConfiguration() .withEnv(env) .endExternalConfiguration() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); List<EnvVar> envs = getContainer(dep).getEnv(); List<EnvVar> selected = envs.stream().filter(var -> var.getName().equals("MY_ENV_VAR")).collect(Collectors.toList()); assertThat(selected.size(), is(1)); assertThat(selected.get(0).getName(), is("MY_ENV_VAR")); assertThat(selected.get(0).getValueFrom().getSecretKeyRef(), is(env.getValueFrom().getSecretKeyRef())); }
@Test public void testExternalConfigurationConfigEnvs() { ExternalConfigurationEnv env = new ExternalConfigurationEnvBuilder() .withName("MY_ENV_VAR") .withNewValueFrom() .withConfigMapKeyRef(new ConfigMapKeySelectorBuilder().withName("my-map").withKey("my-key").withOptional(false).build()) .endValueFrom() .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewExternalConfiguration() .withEnv(env) .endExternalConfiguration() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); List<EnvVar> envs = getContainer(dep).getEnv(); List<EnvVar> selected = envs.stream().filter(var -> var.getName().equals("MY_ENV_VAR")).collect(Collectors.toList()); assertThat(selected.size(), is(1)); assertThat(selected.get(0).getName(), is("MY_ENV_VAR")); assertThat(selected.get(0).getValueFrom().getConfigMapKeyRef(), is(env.getValueFrom().getConfigMapKeyRef())); }
@Test public void testInvalidExternalConfigurationEnvs() { ExternalConfigurationEnv env = new ExternalConfigurationEnvBuilder() .withName("MY_ENV_VAR") .withNewValueFrom() .withConfigMapKeyRef(new ConfigMapKeySelectorBuilder().withName("my-map").withKey("my-key").withOptional(false).build()) .withSecretKeyRef(new SecretKeySelectorBuilder().withName("my-secret").withKey("my-key").withOptional(false).build()) .endValueFrom() .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewExternalConfiguration() .withEnv(env) .endExternalConfiguration() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); List<EnvVar> envs = getContainer(dep).getEnv(); List<EnvVar> selected = envs.stream().filter(var -> var.getName().equals("MY_ENV_VAR")).collect(Collectors.toList()); assertThat(selected.size(), is(0)); }
@Test public void testNoExternalConfigurationEnvs() { ExternalConfigurationEnv env = new ExternalConfigurationEnvBuilder() .withName("MY_ENV_VAR") .withNewValueFrom() .endValueFrom() .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewExternalConfiguration() .withEnv(env) .endExternalConfiguration() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); List<EnvVar> envs = getContainer(dep).getEnv(); List<EnvVar> selected = envs.stream().filter(var -> var.getName().equals("MY_ENV_VAR")).collect(Collectors.toList()); assertThat(selected.size(), is(0)); }
@Test public void testGracePeriod() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewTemplate() .withNewPod() .withTerminationGracePeriodSeconds(123) .endPod() .endTemplate() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); assertThat(dep.getSpec().getTemplate().getSpec().getTerminationGracePeriodSeconds(), is(Long.valueOf(123))); }
@Test public void testDefaultGracePeriod() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource).build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); assertThat(dep.getSpec().getTemplate().getSpec().getTerminationGracePeriodSeconds(), is(Long.valueOf(30))); }
@Test public void testImagePullSecrets() { LocalObjectReference secret1 = new LocalObjectReference("some-pull-secret"); LocalObjectReference secret2 = new LocalObjectReference("some-other-pull-secret"); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewTemplate() .withNewPod() .withImagePullSecrets(secret1, secret2) .endPod() .endTemplate() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); 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 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); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(this.resource, VERSIONS); Deployment dep = kmm2.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"); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewTemplate() .withNewPod() .withImagePullSecrets(secret2) .endPod() .endTemplate() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.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 testDefaultImagePullSecrets() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource).build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); assertThat(dep.getSpec().getTemplate().getSpec().getImagePullSecrets(), is(nullValue())); }
@Test public void testSecurityContext() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewTemplate() .withNewPod() .withSecurityContext(new PodSecurityContextBuilder().withFsGroup(123L).withRunAsGroup(456L).withRunAsUser(789L).build()) .endPod() .endTemplate() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); assertThat(dep.getSpec().getTemplate().getSpec().getSecurityContext(), is(notNullValue())); assertThat(dep.getSpec().getTemplate().getSpec().getSecurityContext().getFsGroup(), is(Long.valueOf(123))); assertThat(dep.getSpec().getTemplate().getSpec().getSecurityContext().getRunAsGroup(), is(Long.valueOf(456))); assertThat(dep.getSpec().getTemplate().getSpec().getSecurityContext().getRunAsUser(), is(Long.valueOf(789))); }
@Test public void testDefaultSecurityContext() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource).build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); assertThat(dep.getSpec().getTemplate().getSpec().getSecurityContext(), is(nullValue())); }
@Test public void testPodDisruptionBudget() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewTemplate() .withNewPodDisruptionBudget() .withMaxUnavailable(2) .endPodDisruptionBudget() .endTemplate() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); PodDisruptionBudget pdb = kmm2.generatePodDisruptionBudget(); assertThat(pdb.getSpec().getMaxUnavailable(), is(new IntOrString(2))); }
@Test public void testDefaultPodDisruptionBudget() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource).build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); PodDisruptionBudget pdb = kmm2.generatePodDisruptionBudget(); assertThat(pdb.getSpec().getMaxUnavailable(), is(new IntOrString(1))); }
@Test public void testImagePullPolicy() { KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(Collections.EMPTY_MAP, true, ImagePullPolicy.ALWAYS, null); assertThat(getContainer(dep).getImagePullPolicy(), is(ImagePullPolicy.ALWAYS.toString())); dep = kmm2.generateDeployment(Collections.EMPTY_MAP, true, ImagePullPolicy.IFNOTPRESENT, null); assertThat(getContainer(dep).getImagePullPolicy(), is(ImagePullPolicy.IFNOTPRESENT.toString())); }
@Test public void testResources() { Map<String, Quantity> requests = new HashMap<>(2); requests.put("cpu", new Quantity("250m")); requests.put("memory", new Quantity("512Mi")); Map<String, Quantity> limits = new HashMap<>(2); limits.put("cpu", new Quantity("500m")); limits.put("memory", new Quantity("1024Mi")); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withResources(new ResourceRequirementsBuilder().withLimits(limits).withRequests(requests).build()) .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(Collections.EMPTY_MAP, true, null, null); Container cont = getContainer(dep); assertThat(cont.getResources().getLimits(), is(limits)); assertThat(cont.getResources().getRequests(), is(requests)); }
@Test public void testJvmOptions() { Map<String, String> xx = new HashMap<>(2); xx.put("UseG1GC", "true"); xx.put("MaxGCPauseMillis", "20"); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewJvmOptions() .withNewXms("512m") .withNewXmx("1024m") .withNewServer(true) .withXx(xx) .endJvmOptions() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(Collections.EMPTY_MAP, true, null, null); Container cont = getContainer(dep); assertThat(cont.getEnv().stream().filter(env -> "KAFKA_JVM_PERFORMANCE_OPTS".equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").contains("-server"), is(true)); assertThat(cont.getEnv().stream().filter(env -> "KAFKA_JVM_PERFORMANCE_OPTS".equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").contains("-XX:+UseG1GC"), is(true)); assertThat(cont.getEnv().stream().filter(env -> "KAFKA_JVM_PERFORMANCE_OPTS".equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").contains("-XX:MaxGCPauseMillis=20"), is(true)); assertThat(cont.getEnv().stream().filter(env -> "KAFKA_HEAP_OPTS".equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").contains("-Xmx1024m"), is(true)); assertThat(cont.getEnv().stream().filter(env -> "KAFKA_HEAP_OPTS".equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").contains("-Xms512m"), is(true)); }
@Test public void testTracing() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withNewJaegerTracing() .endJaegerTracing() .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(Collections.EMPTY_MAP, true, null, null); Container cont = getContainer(dep); assertThat(cont.getEnv().stream().filter(env -> KafkaMirrorMaker2Cluster.ENV_VAR_STRIMZI_TRACING.equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").equals("jaeger"), is(true)); assertThat(cont.getEnv().stream().filter(env -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_CONFIGURATION.equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").contains("consumer.interceptor.classes=io.opentracing.contrib.kafka.TracingConsumerInterceptor"), is(true)); assertThat(cont.getEnv().stream().filter(env -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_CONFIGURATION.equals(env.getName())).map(EnvVar::getValue).findFirst().orElse("").contains("producer.interceptor.classes=io.opentracing.contrib.kafka.TracingProducerInterceptor"), is(true)); }
@Test public void testGenerateDeploymentWithOAuthWithAccessToken() { KafkaMirrorMaker2ClusterSpec targetClusterWithOAuthWithAccessToken = new KafkaMirrorMaker2ClusterSpecBuilder(this.targetCluster) .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withNewAccessToken() .withSecretName("my-token-secret") .withKey("my-token-key") .endAccessToken() .build()) .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withClusters(targetClusterWithOAuthWithAccessToken) .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); Container cont = getContainer(dep); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_SASL_MECHANISM.equals(var.getName())).findFirst().orElse(null).getValue(), is("oauth")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_ACCESS_TOKEN.equals(var.getName())).findFirst().orElse(null).getValueFrom().getSecretKeyRef().getName(), is("my-token-secret")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_ACCESS_TOKEN.equals(var.getName())).findFirst().orElse(null).getValueFrom().getSecretKeyRef().getKey(), is("my-token-key")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_CONFIG.equals(var.getName())).findFirst().orElse(null).getValue().isEmpty(), is(true)); }
@Test public void testGenerateDeploymentWithOAuthWithRefreshToken() { KafkaMirrorMaker2ClusterSpec targetClusterWithOAuthWithRefreshToken = new KafkaMirrorMaker2ClusterSpecBuilder(this.targetCluster) .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .withNewRefreshToken() .withSecretName("my-token-secret") .withKey("my-token-key") .endRefreshToken() .build()) .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withClusters(targetClusterWithOAuthWithRefreshToken) .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); Container cont = getContainer(dep); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_SASL_MECHANISM.equals(var.getName())).findFirst().orElse(null).getValue(), is("oauth")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_REFRESH_TOKEN.equals(var.getName())).findFirst().orElse(null).getValueFrom().getSecretKeyRef().getName(), is("my-token-secret")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_REFRESH_TOKEN.equals(var.getName())).findFirst().orElse(null).getValueFrom().getSecretKeyRef().getKey(), is("my-token-key")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_CONFIG.equals(var.getName())).findFirst().orElse(null).getValue().trim(), is(String.format("%s=\"%s\" %s=\"%s\"", ClientConfig.OAUTH_CLIENT_ID, "my-client-id", ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, "http: }
@Test public void testGenerateDeploymentWithOAuthWithClientSecret() { KafkaMirrorMaker2ClusterSpec targetClusterWithOAuthWithClientSecret = new KafkaMirrorMaker2ClusterSpecBuilder(this.targetCluster) .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .withNewClientSecret() .withSecretName("my-secret-secret") .withKey("my-secret-key") .endClientSecret() .build()) .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withClusters(targetClusterWithOAuthWithClientSecret) .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null); Container cont = getContainer(dep); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_SASL_MECHANISM.equals(var.getName())).findFirst().orElse(null).getValue(), is("oauth")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_CLIENT_SECRET.equals(var.getName())).findFirst().orElse(null).getValueFrom().getSecretKeyRef().getName(), is("my-secret-secret")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_CLIENT_SECRET.equals(var.getName())).findFirst().orElse(null).getValueFrom().getSecretKeyRef().getKey(), is("my-secret-key")); assertThat(cont.getEnv().stream().filter(var -> KafkaMirrorMaker2Cluster.ENV_VAR_KAFKA_CONNECT_OAUTH_CONFIG.equals(var.getName())).findFirst().orElse(null).getValue().trim(), is(String.format("%s=\"%s\" %s=\"%s\"", ClientConfig.OAUTH_CLIENT_ID, "my-client-id", ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, "http: }
@Test public void testGenerateDeploymentWithOAuthWithMissingClientSecret() { assertThrows(InvalidResourceException.class, () -> { KafkaMirrorMaker2ClusterSpec targetClusterWithOAuthWithMissingClientSecret = new KafkaMirrorMaker2ClusterSpecBuilder(this.targetCluster) .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .build()) .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withClusters(targetClusterWithOAuthWithMissingClientSecret) .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithOAuthWithMissingUri() { assertThrows(InvalidResourceException.class, () -> { KafkaMirrorMaker2ClusterSpec targetClusterWithOAuthWithMissingUri = new KafkaMirrorMaker2ClusterSpecBuilder(this.targetCluster) .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withNewClientSecret() .withSecretName("my-secret-secret") .withKey("my-secret-key") .endClientSecret() .build()) .build(); KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withClusters(targetClusterWithOAuthWithMissingUri) .endSpec() .build(); KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithOldVersion() { assertThrows(InvalidResourceException.class, () -> { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .editSpec() .withVersion("2.3.1") .endSpec() .build(); KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS); }); }
@Test public void testNetworkPolicy() { KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource) .build(); KafkaMirrorMaker2Cluster kc = KafkaMirrorMaker2Cluster.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)); }