KafkaBridgeCluster extends AbstractModel { public static KafkaBridgeCluster fromCrd(KafkaBridge kafkaBridge, KafkaVersion.Lookup versions) { KafkaBridgeCluster kafkaBridgeCluster = new KafkaBridgeCluster(kafkaBridge); KafkaBridgeSpec spec = kafkaBridge.getSpec(); kafkaBridgeCluster.tracing = spec.getTracing(); kafkaBridgeCluster.setResources(spec.getResources()); kafkaBridgeCluster.setLogging(spec.getLogging()); kafkaBridgeCluster.setGcLoggingEnabled(spec.getJvmOptions() == null ? DEFAULT_JVM_GC_LOGGING_ENABLED : spec.getJvmOptions().isGcLoggingEnabled()); if (spec.getJvmOptions() != null) { kafkaBridgeCluster.setJavaSystemProperties(spec.getJvmOptions().getJavaSystemProperties()); } String image = spec.getImage(); if (image == null) { image = System.getenv().getOrDefault(ClusterOperatorConfig.STRIMZI_DEFAULT_KAFKA_BRIDGE_IMAGE, "strimzi/kafka-bridge:latest"); } kafkaBridgeCluster.setImage(image); kafkaBridgeCluster.setReplicas(spec.getReplicas()); kafkaBridgeCluster.setBootstrapServers(spec.getBootstrapServers()); kafkaBridgeCluster.setKafkaConsumerConfiguration(spec.getConsumer()); kafkaBridgeCluster.setKafkaProducerConfiguration(spec.getProducer()); if (kafkaBridge.getSpec().getLivenessProbe() != null) { kafkaBridgeCluster.setLivenessProbe(kafkaBridge.getSpec().getLivenessProbe()); } if (kafkaBridge.getSpec().getReadinessProbe() != null) { kafkaBridgeCluster.setReadinessProbe(kafkaBridge.getSpec().getReadinessProbe()); } kafkaBridgeCluster.setMetricsEnabled(spec.getEnableMetrics()); kafkaBridgeCluster.setTls(spec.getTls() != null ? spec.getTls() : null); AuthenticationUtils.validateClientAuthentication(spec.getAuthentication(), spec.getTls() != null); kafkaBridgeCluster.setAuthentication(spec.getAuthentication()); if (spec.getTemplate() != null) { KafkaBridgeTemplate template = spec.getTemplate(); if (template.getDeployment() != null && template.getDeployment().getMetadata() != null) { kafkaBridgeCluster.templateDeploymentLabels = template.getDeployment().getMetadata().getLabels(); kafkaBridgeCluster.templateDeploymentAnnotations = template.getDeployment().getMetadata().getAnnotations(); } ModelUtils.parsePodTemplate(kafkaBridgeCluster, template.getPod()); if (template.getApiService() != null && template.getApiService().getMetadata() != null) { kafkaBridgeCluster.templateServiceLabels = Util.mergeLabelsOrAnnotations(template.getApiService().getMetadata().getLabels(), ModelUtils.getCustomLabelsOrAnnotations(CO_ENV_VAR_CUSTOM_LABELS)); kafkaBridgeCluster.templateServiceAnnotations = template.getApiService().getMetadata().getAnnotations(); } if (template.getBridgeContainer() != null && template.getBridgeContainer().getEnv() != null) { kafkaBridgeCluster.templateContainerEnvVars = template.getBridgeContainer().getEnv(); } if (template.getBridgeContainer() != null && template.getBridgeContainer().getSecurityContext() != null) { kafkaBridgeCluster.templateContainerSecurityContext = template.getBridgeContainer().getSecurityContext(); } ModelUtils.parsePodDisruptionBudgetTemplate(kafkaBridgeCluster, template.getPodDisruptionBudget()); } if (spec.getHttp() != null) { kafkaBridgeCluster.setHttpEnabled(true); kafkaBridgeCluster.setKafkaBridgeHttpConfig(spec.getHttp()); } else { log.warn("No protocol specified."); throw new InvalidResourceException("No protocol for communication with Bridge specified. Use HTTP."); } kafkaBridgeCluster.setOwnerReference(kafkaBridge); return kafkaBridgeCluster; } protected  KafkaBridgeCluster(HasMetadata resource); static KafkaBridgeCluster fromCrd(KafkaBridge kafkaBridge, KafkaVersion.Lookup versions); Service generateService(); Deployment generateDeployment(Map<String, String> annotations, boolean isOpenShift, ImagePullPolicy imagePullPolicy, List<LocalObjectReference> imagePullSecrets); @Override String getAncillaryConfigMapKeyLogConfig(); PodDisruptionBudget generatePodDisruptionBudget(); KafkaBridgeHttpConfig getHttp(); @Override String createLog4jProperties(OrderedProperties properties); static final String APPLICATION_NAME; static final int DEFAULT_REST_API_PORT; static final Probe DEFAULT_HEALTHCHECK_OPTIONS; }
@Test public void testDefaultValues() { KafkaBridgeCluster kbc = KafkaBridgeCluster.fromCrd(ResourceUtils.createEmptyKafkaBridge(namespace, cluster), VERSIONS); assertThat(kbc.image, is("strimzi/kafka-bridge:latest")); assertThat(kbc.replicas, is(KafkaBridgeCluster.DEFAULT_REPLICAS)); assertThat(kbc.readinessProbeOptions.getInitialDelaySeconds(), is(KafkaBridgeCluster.DEFAULT_HEALTHCHECK_DELAY)); assertThat(kbc.readinessProbeOptions.getTimeoutSeconds(), is(KafkaBridgeCluster.DEFAULT_HEALTHCHECK_TIMEOUT)); assertThat(kbc.livenessProbeOptions.getInitialDelaySeconds(), is(KafkaBridgeCluster.DEFAULT_HEALTHCHECK_DELAY)); assertThat(kbc.livenessProbeOptions.getTimeoutSeconds(), is(KafkaBridgeCluster.DEFAULT_HEALTHCHECK_TIMEOUT)); }
@Test public void testFromCrd() { assertThat(kbc.replicas, is(replicas)); assertThat(kbc.image, is(image)); assertThat(kbc.readinessProbeOptions.getInitialDelaySeconds(), is(healthDelay)); assertThat(kbc.readinessProbeOptions.getTimeoutSeconds(), is(healthTimeout)); assertThat(kbc.livenessProbeOptions.getInitialDelaySeconds(), is(healthDelay)); assertThat(kbc.livenessProbeOptions.getTimeoutSeconds(), is(healthTimeout)); }
@Test public void testGenerateDeploymentWithOAuthWithMissingClientSecret() { assertThrows(InvalidResourceException.class, () -> { KafkaBridge resource = new KafkaBridgeBuilder(this.resource) .editSpec() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .build()) .endSpec() .build(); KafkaBridgeCluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithOAuthWithMissingUri() { assertThrows(InvalidResourceException.class, () -> { KafkaBridge resource = new KafkaBridgeBuilder(this.resource) .editSpec() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withNewClientSecret() .withSecretName("my-secret-secret") .withKey("my-secret-key") .endClientSecret() .build()) .endSpec() .build(); KafkaBridgeCluster.fromCrd(resource, VERSIONS); }); }