KafkaCluster extends AbstractModel { public StatefulSet generateStatefulSet(boolean isOpenShift, ImagePullPolicy imagePullPolicy, List<LocalObjectReference> imagePullSecrets) { Map<String, String> stsAnnotations = new HashMap<>(2); stsAnnotations.put(ANNO_STRIMZI_IO_KAFKA_VERSION, kafkaVersion.version()); stsAnnotations.put(ANNO_STRIMZI_IO_STORAGE, ModelUtils.encodeStorageToJson(storage)); return createStatefulSet( stsAnnotations, emptyMap(), getVolumes(isOpenShift), getVolumeClaims(), getMergedAffinity(), getInitContainers(imagePullPolicy), getContainers(imagePullPolicy), imagePullSecrets, isOpenShift); } 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 testGenerateStatefulSet() { StatefulSet sts = kc.generateStatefulSet(true, null, null); checkStatefulSet(sts, kafkaAssembly, true); checkOwnerReference(kc.createOwnerReference(), sts); }
@Test public void withOldAffinityWithoutRack() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withOldAffinityWithoutRack"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getAffinity()); }
@Test public void withAffinityWithoutRack() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withAffinityWithoutRack"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getAffinity()); }
@Test public void withRackWithoutAffinity() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withRackWithoutAffinity"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getAffinity()); }
@Test public void withRackAndOldAffinity() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withRackAndOldAffinity"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getAffinity()); }
@Test public void withRackAndAffinityWithMoreTerms() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withRackAndAffinityWithMoreTerms"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getAffinity()); }
@Test public void withRackAndAffinity() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withRackAndAffinity"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getAffinity()); }
@Test public void withOldTolerations() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withOldTolerations"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getTolerations()); }
@Test public void withTolerations() throws IOException { ResourceTester<Kafka, KafkaCluster> resourceTester = new ResourceTester<>(Kafka.class, VERSIONS, KafkaCluster::fromCrd, this.getClass().getSimpleName() + ".withTolerations"); resourceTester.assertDesiredResource("-STS.yaml", kc -> kc.generateStatefulSet(true, null, null).getSpec().getTemplate().getSpec().getTolerations()); }