FIx KAAP Issues

migration-tool/src/main/java/com/datastax/oss/kaap/migrationtool/PulsarClusterResourceGenerator.java

@@ -44,6 +44,7 @@
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.BooleanUtils;
 import org.apache.commons.lang3.StringUtils;
 
 @Slf4j
@@ -210,11 +211,11 @@ private AuthConfig getAuthConfig() {
                         )
                         .contains(spec.getSpecName()) && spec.isEnabled();
 
-        final boolean authEnabled = getValueAssertSame(
+        final boolean authEnabled = BooleanUtils.isTrue(getValueAssertSame(
                 filterComponents,
                 spec -> parseConfigValueBool(spec.getConfig(), "authenticationEnabled"),
                 false,
-                "authenticationEnabled").booleanValue();
+                "authenticationEnabled"));
         if (!authEnabled) {
             log.info("Found authenticationEnabled=false, auth will be disabled in the CRD");
             return AuthConfig.builder()

migration-tool/src/main/java/com/datastax/oss/kaap/migrationtool/specs/AutorecoverySpecGenerator.java

@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.Objects;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 
 @Slf4j
 public class AutorecoverySpecGenerator extends BaseSpecGenerator<AutorecoverySpec> {
@@ -82,8 +83,9 @@ public void internalGenerateSpec() {
         isRestartOnConfigMapChange = isPodDependantOnConfigMap(deploymentSpec.getTemplate());
         priorityClassName = spec.getPriorityClassName();
         config = convertConfigMapData(configMap);
-        if (!config.containsKey("ensemblePlacementPolicy")) {
-            config.put("ensemblePlacementPolicy", "");
+        Object ensemblePlacementPolicy = config.get("ensemblePlacementPolicy");
+        if (ensemblePlacementPolicy instanceof String policy && StringUtils.isBlank(policy)) {
+            config.remove("ensemblePlacementPolicy");
         }
         tlsEntryConfig = createTlsEntryConfig(deployment);
 

operator/src/main/java/com/datastax/oss/kaap/controllers/bookkeeper/BookKeeperResourcesFactory.java

@@ -295,9 +295,9 @@ public StatefulSet generateStatefulSet() {
             );
         } else {
             persistentVolumeClaims.add(createPersistentVolumeClaim(journalVolumeName, spec.getVolumes().getJournal(),
-                    labels));
+                    Map.of()));
             persistentVolumeClaims.add(createPersistentVolumeClaim(ledgersVolumeName, spec.getVolumes().getLedgers(),
-                    labels));
+                    Map.of()));
         }
 
 

operator/src/main/java/com/datastax/oss/kaap/controllers/zookeeper/ZooKeeperResourcesFactory.java

@@ -331,7 +331,7 @@ public void patchStatefulSet() {
         } else {
             final VolumeConfig dataVolume = spec.getDataVolume();
             persistentVolumeClaims.add(
-                    createPersistentVolumeClaim(dataStorageVolumeName, dataVolume, labels)
+                    createPersistentVolumeClaim(dataStorageVolumeName, dataVolume, Map.of())
             );
         }