KafkaMirrorMakerCluster extends AbstractModel { public static KafkaMirrorMakerCluster fromCrd(KafkaMirrorMaker kafkaMirrorMaker, KafkaVersion.Lookup versions) { KafkaMirrorMakerCluster kafkaMirrorMakerCluster = new KafkaMirrorMakerCluster(kafkaMirrorMaker); KafkaMirrorMakerSpec spec = kafkaMirrorMaker.getSpec(); if (spec != null) { kafkaMirrorMakerCluster.setReplicas(spec.getReplicas()); kafkaMirrorMakerCluster.setResources(spec.getResources()); if (spec.getReadinessProbe() != null) { kafkaMirrorMakerCluster.setReadinessProbe(spec.getReadinessProbe()); } if (spec.getLivenessProbe() != null) { kafkaMirrorMakerCluster.setLivenessProbe(spec.getLivenessProbe()); } kafkaMirrorMakerCluster.setWhitelist(spec.getWhitelist()); AuthenticationUtils.validateClientAuthentication(spec.getProducer().getAuthentication(), spec.getProducer().getTls() != null); kafkaMirrorMakerCluster.setProducer(spec.getProducer()); AuthenticationUtils.validateClientAuthentication(spec.getConsumer().getAuthentication(), spec.getConsumer().getTls() != null); kafkaMirrorMakerCluster.setConsumer(spec.getConsumer()); kafkaMirrorMakerCluster.setImage(versions.kafkaMirrorMakerImage(spec.getImage(), spec.getVersion())); kafkaMirrorMakerCluster.setLogging(spec.getLogging()); kafkaMirrorMakerCluster.setGcLoggingEnabled(spec.getJvmOptions() == null ? DEFAULT_JVM_GC_LOGGING_ENABLED : spec.getJvmOptions().isGcLoggingEnabled()); if (spec.getJvmOptions() != null) { kafkaMirrorMakerCluster.setJavaSystemProperties(spec.getJvmOptions().getJavaSystemProperties()); } kafkaMirrorMakerCluster.setJvmOptions(spec.getJvmOptions()); Map<String, Object> metrics = spec.getMetrics(); if (metrics != null) { kafkaMirrorMakerCluster.setMetricsEnabled(true); kafkaMirrorMakerCluster.setMetricsConfig(metrics.entrySet()); } if (spec.getTemplate() != null) { KafkaMirrorMakerTemplate template = spec.getTemplate(); if (template.getDeployment() != null && template.getDeployment().getMetadata() != null) { kafkaMirrorMakerCluster.templateDeploymentLabels = template.getDeployment().getMetadata().getLabels(); kafkaMirrorMakerCluster.templateDeploymentAnnotations = template.getDeployment().getMetadata().getAnnotations(); } if (template.getMirrorMakerContainer() != null && template.getMirrorMakerContainer().getEnv() != null) { kafkaMirrorMakerCluster.templateContainerEnvVars = template.getMirrorMakerContainer().getEnv(); } if (template.getMirrorMakerContainer() != null && template.getMirrorMakerContainer().getSecurityContext() != null) { kafkaMirrorMakerCluster.templateContainerSecurityContext = template.getMirrorMakerContainer().getSecurityContext(); } ModelUtils.parsePodTemplate(kafkaMirrorMakerCluster, template.getPod()); ModelUtils.parsePodDisruptionBudgetTemplate(kafkaMirrorMakerCluster, template.getPodDisruptionBudget()); } kafkaMirrorMakerCluster.setUserAffinity(affinity(spec)); kafkaMirrorMakerCluster.setTolerations(tolerations(spec)); kafkaMirrorMakerCluster.tracing = spec.getTracing(); } kafkaMirrorMakerCluster.setOwnerReference(kafkaMirrorMaker); return kafkaMirrorMakerCluster; } protected  KafkaMirrorMakerCluster(HasMetadata resource); static KafkaMirrorMakerCluster fromCrd(KafkaMirrorMaker kafkaMirrorMaker, KafkaVersion.Lookup versions); Deployment generateDeployment(Map<String, String> annotations, boolean isOpenShift, ImagePullPolicy imagePullPolicy, List<LocalObjectReference> imagePullSecrets); PodDisruptionBudget generatePodDisruptionBudget(); void setWhitelist(String whitelist); void setProducer(KafkaMirrorMakerProducerSpec producer); void setConsumer(KafkaMirrorMakerConsumerSpec consumer); static final Probe READINESS_PROBE_OPTIONS; }
@Test public void testDefaultValues() { KafkaMirrorMakerProducerSpec producer = new KafkaMirrorMakerProducerSpecBuilder() .withBootstrapServers(producerBootstrapServers) .build(); KafkaMirrorMakerConsumerSpec consumer = new KafkaMirrorMakerConsumerSpecBuilder() .withBootstrapServers(consumerBootstrapServers) .withGroupId(groupId) .withNumStreams(numStreams) .build(); KafkaMirrorMaker resource = new KafkaMirrorMakerBuilder(ResourceUtils.createEmptyKafkaMirrorMaker(namespace, cluster)) .withNewSpec() .withReplicas(replicas) .withProducer(producer) .withConsumer(consumer) .endSpec() .build(); KafkaMirrorMakerCluster mm = KafkaMirrorMakerCluster.fromCrd(resource, VERSIONS); assertThat(mm.image, is(KafkaVersionTestUtils.DEFAULT_KAFKA_MIRROR_MAKER_IMAGE)); assertThat(new KafkaMirrorMakerConsumerConfiguration(mm.consumer.getConfig().entrySet()).getConfiguration(), is(defaultConsumerConfiguration)); assertThat(new KafkaMirrorMakerProducerConfiguration(mm.producer.getConfig().entrySet()).getConfiguration(), is(defaultProducerConfiguration)); }
@Test public void testFromCrd() { assertThat(mm.getReplicas(), is(replicas)); assertThat(mm.getImage(), is(image)); assertThat(mm.consumer.getBootstrapServers(), is(consumerBootstrapServers)); assertThat(mm.producer.getBootstrapServers(), is(producerBootstrapServers)); assertThat(mm.getWhitelist(), is(whitelist)); assertThat(mm.consumer.getGroupId(), is(groupId)); }
@Test public void testGenerateDeploymentWithConsumerOAuthWithMissingClientSecret() { assertThrows(InvalidResourceException.class, () -> { KafkaMirrorMaker resource = new KafkaMirrorMakerBuilder(this.resource) .editSpec() .editConsumer() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .build()) .endConsumer() .endSpec() .build(); KafkaMirrorMakerCluster kc = KafkaMirrorMakerCluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithConsumerOAuthWithMissingUri() { assertThrows(InvalidResourceException.class, () -> { KafkaMirrorMaker resource = new KafkaMirrorMakerBuilder(this.resource) .editSpec() .editConsumer() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withNewClientSecret() .withSecretName("my-secret-secret") .withKey("my-secret-key") .endClientSecret() .build()) .endConsumer() .endSpec() .build(); KafkaMirrorMakerCluster kc = KafkaMirrorMakerCluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithProducerOAuthWithMissingClientSecret() { assertThrows(InvalidResourceException.class, () -> { KafkaMirrorMaker resource = new KafkaMirrorMakerBuilder(this.resource) .editSpec() .editProducer() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .build()) .endProducer() .endSpec() .build(); KafkaMirrorMakerCluster kc = KafkaMirrorMakerCluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithProducerOAuthWithMissingUri() { assertThrows(InvalidResourceException.class, () -> { KafkaMirrorMaker resource = new KafkaMirrorMakerBuilder(this.resource) .editSpec() .editProducer() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withNewClientSecret() .withSecretName("my-secret-secret") .withKey("my-secret-key") .endClientSecret() .build()) .endProducer() .endSpec() .build(); KafkaMirrorMakerCluster kc = KafkaMirrorMakerCluster.fromCrd(resource, VERSIONS); }); }