KafkaCluster extends AbstractModel { public static KafkaCluster fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions) { return fromCrd(kafkaAssembly, versions, null, 0); } private  KafkaCluster(HasMetadata resource); static String kafkaClusterName(String cluster); static String metricAndLogConfigsName(String cluster); static String serviceName(String cluster); static String podDnsName(String namespace, String cluster, int podId); static String podDnsName(String namespace, String cluster, String podName); static String podDnsNameWithoutClusterDomain(String namespace, String cluster, int podId); static String podDnsNameWithoutClusterDomain(String namespace, String cluster, String podName); static String externalBootstrapServiceName(String cluster); static String externalServiceName(String cluster, int pod); static String headlessServiceName(String cluster); static String kafkaPodName(String cluster, int pod); static String clientsCaKeySecretName(String cluster); static String brokersSecretName(String cluster); static String jmxSecretName(String cluster); static String clientsCaCertSecretName(String cluster); static KafkaCluster fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions); @SuppressWarnings({"checkstyle:MethodLength", "checkstyle:JavaNCSS", "deprecation"}) static KafkaCluster fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions, Storage oldStorage, int oldReplicas); void generateCertificates(Kafka kafka, ClusterCa clusterCa, Set<String> externalBootstrapDnsName,
            Map<Integer, Set<String>> externalDnsNames, boolean isMaintenanceTimeWindowsSatisfied); Service generateService(); List<Service> generateExternalBootstrapServices(); List<Service> generateExternalServices(int pod); List<Route> generateExternalBootstrapRoutes(); List<Route> generateExternalRoutes(int pod); List<Ingress> generateExternalBootstrapIngresses(); List<Ingress> generateExternalIngresses(int pod); Service generateHeadlessService(); StatefulSet generateStatefulSet(boolean isOpenShift,
                                           ImagePullPolicy imagePullPolicy,
                                           List<LocalObjectReference> imagePullSecrets); Secret generateBrokersSecret(); Secret generateJmxSecret(); int getClientPort(); int getClientTlsPort(); int getLoadbalancerPort(); int getRoutePort(); int getIngressPort(); List<PersistentVolumeClaim> generatePersistentVolumeClaims(Storage storage); @Override String getServiceAccountName(); ClusterRoleBinding generateClusterRoleBinding(String assemblyNamespace); static String policyName(String cluster); NetworkPolicy generateNetworkPolicy(boolean namespaceAndPodSelectorNetworkPolicySupported); PodDisruptionBudget generatePodDisruptionBudget(); void setListeners(List<GenericKafkaListener> listeners); List<GenericKafkaListener> getListeners(); boolean isJmxEnabled(); void setJmxEnabled(boolean jmxEnabled); void setAuthorization(KafkaAuthorization authorization); boolean isExposed(); boolean isExposedWithRoute(); boolean isExposedWithLoadBalancer(); boolean isExposedWithNodePort(); boolean isExposedWithIngress(); String getAdvertisedHostname(GenericKafkaListener listener, int podNumber, String address); String getAdvertisedPort(GenericKafkaListener listener, int podNumber, Integer port); @Override KafkaConfiguration getConfiguration(); boolean isJmxAuthenticated(); void setJmxAuthenticated(boolean jmxAuthenticated); NodeAddressType getPreferredNodeAddressType(GenericKafkaListener listener); String getBrokersConfiguration(); ConfigMap generateAncillaryConfigMap(ConfigMap externalLoggingCm, Set<String> advertisedHostnames, Set<String> advertisedPorts); KafkaVersion getKafkaVersion(); static final int REPLICATION_PORT; static final String ANNO_STRIMZI_IO_KAFKA_VERSION; static final String ANNO_STRIMZI_IO_FROM_VERSION; static final String ANNO_STRIMZI_IO_TO_VERSION; static final String ANNO_STRIMZI_BROKER_CONFIGURATION_HASH; static final String ANNO_STRIMZI_CUSTOM_LISTENER_CERT_THUMBPRINTS; static final String BROKER_CONFIGURATION_FILENAME; static final String BROKER_LISTENERS_FILENAME; static final String BROKER_ADVERTISED_HOSTNAMES_FILENAME; static final String BROKER_ADVERTISED_PORTS_FILENAME; static final Probe DEFAULT_HEALTHCHECK_OPTIONS; }
@Test public void testGeneratePersistentVolumeClaimsJbodWithoutVolumes() { assertThrows(InvalidResourceException.class, () -> { Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withStorage(new JbodStorageBuilder().withVolumes(Collections.EMPTY_LIST) .build()) .endKafka() .endSpec() .build(); KafkaCluster.fromCrd(kafkaAssembly, VERSIONS); }); }
@Test public void testStorageValidationAfterInitialDeployment() { assertThrows(InvalidResourceException.class, () -> { Storage oldStorage = new JbodStorageBuilder() .withVolumes(new PersistentClaimStorageBuilder().withSize("100Gi").build()) .build(); Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withStorage(new JbodStorageBuilder().withVolumes(Collections.EMPTY_LIST) .build()) .endKafka() .endSpec() .build(); KafkaCluster.fromCrd(kafkaAssembly, VERSIONS, oldStorage, replicas); }); }
@Test public void testStorageReverting() { Storage jbod = new JbodStorageBuilder().withVolumes( new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withDeleteClaim(false).withId(0).withSize("100Gi").build(), new PersistentClaimStorageBuilder().withStorageClass("gp2-st1").withDeleteClaim(true).withId(1).withSize("1000Gi").build()) .build(); Storage ephemeral = new EphemeralStorageBuilder().build(); Storage persistent = new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withDeleteClaim(false).withId(0).withSize("100Gi").build(); Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withStorage(jbod) .endKafka() .endSpec() .build(); KafkaCluster kc = KafkaCluster.fromCrd(kafkaAssembly, VERSIONS, ephemeral, replicas); assertThat(kc.getStorage(), is(ephemeral)); assertThat(kc.getWarningConditions().size(), is(1)); assertThat(kc.getWarningConditions().get(0).getReason(), is("KafkaStorage")); kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withStorage(jbod) .endKafka() .endSpec() .build(); kc = KafkaCluster.fromCrd(kafkaAssembly, VERSIONS, persistent, replicas); assertThat(kc.getStorage(), is(persistent)); assertThat(kc.getWarningConditions().size(), is(1)); assertThat(kc.getWarningConditions().get(0).getReason(), is("KafkaStorage")); kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withStorage(ephemeral) .endKafka() .endSpec() .build(); kc = KafkaCluster.fromCrd(kafkaAssembly, VERSIONS, jbod, replicas); assertThat(kc.getStorage(), is(jbod)); assertThat(kc.getWarningConditions().size(), is(1)); assertThat(kc.getWarningConditions().get(0).getReason(), is("KafkaStorage")); kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withStorage(persistent) .endKafka() .endSpec() .build(); kc = KafkaCluster.fromCrd(kafkaAssembly, VERSIONS, jbod, replicas); assertThat(kc.getStorage(), is(jbod)); assertThat(kc.getWarningConditions().size(), is(1)); assertThat(kc.getWarningConditions().get(0).getReason(), is("KafkaStorage")); }
@Test public void testExternalIngressMissingConfiguration() { GenericKafkaListenerConfigurationBroker broker0 = new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .build(); Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withNewListeners() .addNewGenericKafkaListener() .withName("external") .withPort(9094) .withType(KafkaListenerType.INGRESS) .withTls(true) .withNewConfiguration() .withNewIngressClass("nginx-internal") .withNewBootstrap() .withHost("my-kafka-bootstrap.com") .withAnnotations(Collections.singletonMap("dns-annotation", "my-kafka-bootstrap.com")) .endBootstrap() .withBrokers(broker0) .endConfiguration() .endGenericKafkaListener() .endListeners() .endKafka() .endSpec() .build(); try { KafkaCluster kc = KafkaCluster.fromCrd(kafkaAssembly, VERSIONS); fail("Expected exception was not thrown"); } catch (InvalidResourceException e) { } }
@Test public void testGenerateDeploymentWithKeycloakAuthorizationMissingOAuthListeners() { assertThrows(InvalidResourceException.class, () -> { Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withAuthorization( new KafkaAuthorizationKeycloakBuilder() .build()) .endKafka() .endSpec() .build(); KafkaCluster.fromCrd(kafkaAssembly, VERSIONS); }); }
@Test public void testCruiseControlWithSingleNodeKafka() { Map<String, Object> config = new HashMap<>(); config.put("offsets.topic.replication.factor", 1); config.put("transaction.state.log.replication.factor", 1); config.put("transaction.state.log.min.isr", 1); Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withReplicas(1) .withConfig(config) .endKafka() .withNewCruiseControl() .endCruiseControl() .endSpec() .build(); InvalidResourceException ex = assertThrows(InvalidResourceException.class, () -> { KafkaCluster.fromCrd(kafkaAssembly, VERSIONS); }); assertThat(ex.getMessage(), is("Kafka " + namespace + "/" + cluster + " has invalid configuration. " + "Cruise Control cannot be deployed with a single-node Kafka cluster. It requires at least two Kafka nodes.")); }