KafkaCluster extends AbstractModel { @Override protected List<EnvVar> getEnvVars() { List<EnvVar> varList = new ArrayList<>(); varList.add(buildEnvVar(ENV_VAR_KAFKA_METRICS_ENABLED, String.valueOf(isMetricsEnabled))); varList.add(buildEnvVar(ENV_VAR_STRIMZI_KAFKA_GC_LOG_ENABLED, String.valueOf(gcLoggingEnabled))); if (javaSystemProperties != null) { varList.add(buildEnvVar(ENV_VAR_STRIMZI_JAVA_SYSTEM_PROPERTIES, ModelUtils.getJavaSystemPropertiesToString(javaSystemProperties))); } heapOptions(varList, 0.5, 5L * 1024L * 1024L * 1024L); jvmPerformanceOptions(varList); for (GenericKafkaListener listener : listeners) { if (listener.getAuth() != null && listener.getAuth() instanceof KafkaListenerAuthenticationOAuth) { KafkaListenerAuthenticationOAuth oauth = (KafkaListenerAuthenticationOAuth) listener.getAuth(); if (oauth.getClientSecret() != null) { varList.add(buildEnvVarFromSecret("STRIMZI_" + ListenersUtils.envVarIdentifier(listener) + "_OAUTH_CLIENT_SECRET", oauth.getClientSecret().getSecretName(), oauth.getClientSecret().getKey())); } } } if (isJmxEnabled()) { varList.add(buildEnvVar(ENV_VAR_KAFKA_JMX_ENABLED, "true")); if (isJmxAuthenticated) { varList.add(buildEnvVarFromSecret(ENV_VAR_KAFKA_JMX_USERNAME, jmxSecretName(cluster), SECRET_JMX_USERNAME_KEY)); varList.add(buildEnvVarFromSecret(ENV_VAR_KAFKA_JMX_PASSWORD, jmxSecretName(cluster), SECRET_JMX_PASSWORD_KEY)); } } varList.addAll(getRequiredEnvVars()); addContainerEnvsToExistingEnvs(varList, templateKafkaContainerEnvVars); return varList; } 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 testJavaSystemProperties() { assertThat(kc.getEnvVars().get(2).getName(), is("STRIMZI_JAVA_SYSTEM_PROPERTIES")); assertThat(kc.getEnvVars().get(2).getValue(), is("-D" + javaSystemProperties.get(0).getName() + "=" + javaSystemProperties.get(0).getValue() + " " + "-D" + javaSystemProperties.get(1).getName() + "=" + javaSystemProperties.get(1).getValue())); }