KafkaConnectCluster extends AbstractModel { public static KafkaConnectCluster fromCrd(KafkaConnect kafkaConnect, KafkaVersion.Lookup versions) { KafkaConnectCluster cluster = fromSpec(kafkaConnect.getSpec(), versions, new KafkaConnectCluster(kafkaConnect)); cluster.setOwnerReference(kafkaConnect); return cluster; } protected  KafkaConnectCluster(HasMetadata resource); protected  KafkaConnectCluster(HasMetadata resource, String applicationName); static KafkaConnectCluster fromCrd(KafkaConnect kafkaConnect, KafkaVersion.Lookup versions); Service generateService(); Deployment generateDeployment(Map<String, String> annotations, boolean isOpenShift, ImagePullPolicy imagePullPolicy, List<LocalObjectReference> imagePullSecrets); PodDisruptionBudget generatePodDisruptionBudget(); @Override String getServiceAccountName(); NetworkPolicy generateNetworkPolicy(boolean namespaceAndPodSelectorNetworkPolicySupported, boolean connectorOperatorEnabled); Tracing getTracing(); ClusterRoleBinding generateClusterRoleBinding(); static final int REST_API_PORT; static final Probe DEFAULT_HEALTHCHECK_OPTIONS; }
@Test public void testDefaultValues() { KafkaConnectCluster kc = KafkaConnectCluster.fromCrd(ResourceUtils.createEmptyKafkaConnect(namespace, cluster), VERSIONS); assertThat(kc.image, is(KafkaVersionTestUtils.DEFAULT_KAFKA_CONNECT_IMAGE)); assertThat(kc.replicas, is(KafkaConnectCluster.DEFAULT_REPLICAS)); assertThat(kc.readinessProbeOptions.getInitialDelaySeconds(), is(KafkaConnectCluster.DEFAULT_HEALTHCHECK_DELAY)); assertThat(kc.readinessProbeOptions.getTimeoutSeconds(), is(KafkaConnectCluster.DEFAULT_HEALTHCHECK_TIMEOUT)); assertThat(kc.livenessProbeOptions.getInitialDelaySeconds(), is(KafkaConnectCluster.DEFAULT_HEALTHCHECK_DELAY)); assertThat(kc.livenessProbeOptions.getTimeoutSeconds(), is(KafkaConnectCluster.DEFAULT_HEALTHCHECK_TIMEOUT)); assertThat(kc.getConfiguration().asOrderedProperties(), is(defaultConfiguration)); }
@Test public void testFromCrd() { assertThat(kc.replicas, is(replicas)); assertThat(kc.image, 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)); }
@Test public void testGenerateDeploymentWithOAuthWithMissingClientSecret() { assertThrows(InvalidResourceException.class, () -> { KafkaConnect resource = new KafkaConnectBuilder(this.resource) .editSpec() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withTokenEndpointUri("http: .build()) .endSpec() .build(); KafkaConnectCluster kc = KafkaConnectCluster.fromCrd(resource, VERSIONS); }); }
@Test public void testGenerateDeploymentWithOAuthWithMissingUri() { assertThrows(InvalidResourceException.class, () -> { KafkaConnect resource = new KafkaConnectBuilder(this.resource) .editSpec() .withAuthentication( new KafkaClientAuthenticationOAuthBuilder() .withClientId("my-client-id") .withNewClientSecret() .withSecretName("my-secret-secret") .withKey("my-secret-key") .endClientSecret() .build()) .endSpec() .build(); KafkaConnectCluster kc = KafkaConnectCluster.fromCrd(resource, VERSIONS); }); }