KafkaCluster extends AbstractModel { protected static void validateIntConfigProperty(String propertyName, KafkaClusterSpec kafkaClusterSpec) { String orLess = kafkaClusterSpec.getReplicas() > 1 ? " or less" : ""; if (kafkaClusterSpec.getConfig() != null && kafkaClusterSpec.getConfig().get(propertyName) != null) { try { int propertyVal = Integer.parseInt(kafkaClusterSpec.getConfig().get(propertyName).toString()); if (propertyVal > kafkaClusterSpec.getReplicas()) { throw new InvalidResourceException("Kafka configuration option '" + propertyName + "' should be set to " + kafkaClusterSpec.getReplicas() + orLess + " because 'spec.kafka.replicas' is " + kafkaClusterSpec.getReplicas()); } } catch (NumberFormatException e) { throw new InvalidResourceException("Property " + propertyName + " should be an integer"); } } } 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 testReplicasAndRelatedOptionsValidationNok() { String propertyName = "offsets.topic.replication.factor"; Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withConfig(singletonMap(propertyName, replicas + 1)) .endKafka() .endSpec() .build(); InvalidResourceException ex = assertThrows(InvalidResourceException.class, () -> { KafkaCluster.validateIntConfigProperty(propertyName, kafkaAssembly.getSpec().getKafka()); }); assertThat(ex.getMessage().equals("Kafka configuration option '" + propertyName + "' should be set to " + replicas + " or less because 'spec.kafka.replicas' is " + replicas), is(true)); }
@Test public void testReplicasAndRelatedOptionsValidationOk() { Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCm, configuration, emptyMap())) .editSpec() .editKafka() .withConfig(singletonMap("offsets.topic.replication.factor", replicas - 1)) .endKafka() .endSpec() .build(); KafkaCluster.validateIntConfigProperty("offsets.topic.replication.factor", kafkaAssembly.getSpec().getKafka()); }