ZookeeperCluster extends AbstractModel { public static ZookeeperCluster fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions) { return fromCrd(kafkaAssembly, versions, null, 0); } private  ZookeeperCluster(HasMetadata resource); static String zookeeperClusterName(String cluster); static String zookeeperMetricAndLogConfigsName(String cluster); static String serviceName(String cluster); static String headlessServiceName(String cluster); static String podDnsName(String namespace, String cluster, int podId); static String podDnsNameWithoutSuffix(String namespace, String cluster, int podId); static String zookeeperPodName(String cluster, int pod); static String nodesSecretName(String cluster); static ZookeeperCluster fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions); @SuppressWarnings({"checkstyle:MethodLength", "checkstyle:CyclomaticComplexity"}) static ZookeeperCluster fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions, Storage oldStorage, int oldReplicas); Service generateService(); static String policyName(String cluster); NetworkPolicy generateNetworkPolicy(boolean namespaceAndPodSelectorNetworkPolicySupported); Service generateHeadlessService(); StatefulSet generateStatefulSet(boolean isOpenShift, ImagePullPolicy imagePullPolicy, List<LocalObjectReference> imagePullSecrets); Secret generateNodesSecret(ClusterCa clusterCa, Kafka kafka, boolean isMaintenanceTimeWindowsSatisfied); List<PersistentVolumeClaim> generatePersistentVolumeClaims(); PodDisruptionBudget generatePodDisruptionBudget(); static String containerServiceAccountName(String zookeeperResourceName); @Override void setImage(String image); void setVersion(String version); String getVersion(); ConfigMap generateConfigurationConfigMap(ConfigMap cm); static final int CLIENT_PLAINTEXT_PORT; static final int CLIENT_TLS_PORT; static final int CLUSTERING_PORT; static final int LEADER_ELECTION_PORT; static final String ZOOKEEPER_NAME; static final Probe DEFAULT_HEALTHCHECK_OPTIONS; static final String ENV_VAR_ZOOKEEPER_METRICS_ENABLED; static final String ENV_VAR_ZOOKEEPER_CONFIGURATION; static final String ENV_VAR_ZOOKEEPER_SNAPSHOT_CHECK_ENABLED; static final String CONFIG_MAP_KEY_ZOOKEEPER_NODE_COUNT; }
@Test public void testInvalidVersion() { assertThrows(InvalidResourceException.class, () -> { Kafka ka = new KafkaBuilder(this.ka) .editSpec() .editKafka() .withImage(null) .withVersion("10000.0.0") .endKafka() .editZookeeper() .withImage(null) .endZookeeper() .endSpec() .build(); ZookeeperCluster.fromCrd(ka, VERSIONS); }); }
@Test public void testStorageReverting() { SingleVolumeStorage ephemeral = new EphemeralStorageBuilder().build(); SingleVolumeStorage persistent = new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withDeleteClaim(false).withId(0).withSize("100Gi").build(); Kafka ka = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCmJson, configurationJson, zooConfigurationJson)) .editSpec() .editZookeeper() .withStorage(ephemeral) .endZookeeper() .endSpec() .build(); ZookeeperCluster zc = ZookeeperCluster.fromCrd(ka, VERSIONS, persistent, replicas); assertThat(zc.getStorage(), is(persistent)); ka = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCmJson, configurationJson, zooConfigurationJson)) .editSpec() .editZookeeper() .withStorage(persistent) .endZookeeper() .endSpec() .build(); zc = ZookeeperCluster.fromCrd(ka, VERSIONS, ephemeral, replicas); assertThat(zc.getStorage(), is(ephemeral)); assertThat(zc.getWarningConditions().size(), is(1)); assertThat(zc.getWarningConditions().get(0).getReason(), is("ZooKeeperStorage")); }
@Test public void testStorageValidationAfterInitialDeployment() { assertThrows(InvalidResourceException.class, () -> { Storage oldStorage = new PersistentClaimStorageBuilder() .withSize("100Gi") .build(); Kafka kafkaAssembly = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout, metricsCmJson, configurationJson, zooConfigurationJson)) .editSpec() .editZookeeper() .withStorage(new PersistentClaimStorageBuilder().build()) .endZookeeper() .endSpec() .build(); ZookeeperCluster.fromCrd(kafkaAssembly, VERSIONS, oldStorage, replicas); }); }