Configures entry_location_rocksdb.conf

operator/src/main/java/com/datastax/oss/kaap/controllers/BaseResourcesFactory.java

@@ -79,6 +79,8 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import lombok.extern.jbosslog.JBossLog;
 import org.apache.commons.codec.digest.DigestUtils;
@@ -88,6 +90,15 @@
 @JBossLog
 public abstract class BaseResourcesFactory<T> {
 
+    // Regex to capture the version string after a colon and optionally extract major version
+    // This pattern looks for a colon, then captures the entire version string
+    // The first group (\\d+) specifically captures the major version
+    public static final Pattern IMAGE_STRING_PATTERN = Pattern.compile(
+        ":(?<fullVersion>(\\d+)(?:\\.\\d+){0,2}(?:[\\-][a-zA-Z0-9\\.]+)?)");
+
+    public static final Pattern MAJOR_VERSION_PATTERN = Pattern.compile("^(\\d+)");
+
+
     public static final String CONFIG_PULSAR_PREFIX = "PULSAR_PREFIX_";
     public static final String DEPLOYMENT_REVISION_ANNOTATION = "deployment.kubernetes.io/revision";
     protected final KubernetesClient client;
@@ -1137,4 +1148,25 @@ protected static List<Container> getSidecars(List<Container> containers) {
         }
         return containers;
     }
+
+    public int getPulsarMajorVersion(String imageString) {
+        int pulsarMajorVersion = 2;
+        if (imageString == null) {
+            return pulsarMajorVersion;
+        }
+        Matcher imageStringMatcher = IMAGE_STRING_PATTERN.matcher(imageString);
+
+        if (imageStringMatcher.find()) {
+            String fullVersion = imageStringMatcher.group("fullVersion");
+            System.out.println("Full Version: " + fullVersion);
+            // Extract the major version specifically from the fullVersion string
+            Matcher majorVersionMatcher = MAJOR_VERSION_PATTERN.matcher(fullVersion);
+            if (majorVersionMatcher.find()) {
+                String majorVersionStr = majorVersionMatcher.group(1);
+                System.out.println("Major Version: " + majorVersionStr);
+                pulsarMajorVersion = Integer.parseInt(majorVersionStr);
+            }
+        }
+        return pulsarMajorVersion;
+    }
 }

operator/src/main/java/com/datastax/oss/kaap/controllers/autorecovery/AutorecoveryResourcesFactory.java

@@ -140,7 +140,15 @@ public void patchDeployment() {
             addTlsVolumes(volumeMounts, volumes, getTlsSecretNameForAutorecovery());
         }
         String mainArg = "bin/apply-config-from-env.py conf/bookkeeper.conf && ";
-        if (tlsEnabledOnBookKeeper) {
+
+        int pulsarMajorVersion = getPulsarMajorVersion(spec.getImage());
+        if (pulsarMajorVersion >= 4) {
+            mainArg += "bin/update-ini-from-env.py conf/entry_location_rocksdb.conf PULSAR_PREFIX_entry_location_rocksdb_ && ";
+        } else if (pulsarMajorVersion >= 3) {
+            mainArg += "bin/apply-config-from-env.py conf/entry_location_rocksdb.conf --prefix PULSAR_PREFIX_entry_location_rocksdb_TableOptions_ && ";
+        }
+
+      if (tlsEnabledOnBookKeeper) {
             mainArg += "openssl pkcs8 -topk8 -inform PEM -outform PEM -in /pulsar/certs/tls.key "
                     + "-out /pulsar/tls-pk8.key -nocrypt && ";
         }

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

@@ -59,6 +59,7 @@ public class BookKeeperResourcesFactory extends BaseResourcesFactory<BookKeeperS
     public static final int DEFAULT_BK_PORT = 3181;
     public static final int DEFAULT_HTTP_PORT = 8000;
 
+
     public static List<String> getInitContainerNames(String clusterName, String baseName) {
         return List.of(getMainContainerName(clusterName, baseName));
     }
@@ -247,6 +248,14 @@ public StatefulSet generateStatefulSet() {
         );
 
         String mainArg = "bin/apply-config-from-env.py conf/bookkeeper.conf && ";
+
+        int pulsarMajorVersion = getPulsarMajorVersion(spec.getImage());
+        if (pulsarMajorVersion >= 4) {
+            mainArg += "bin/update-ini-from-env.py conf/entry_location_rocksdb.conf PULSAR_PREFIX_entry_location_rocksdb_ && ";
+        } else if (pulsarMajorVersion >= 3) {
+            mainArg += "bin/apply-config-from-env.py conf/entry_location_rocksdb.conf --prefix PULSAR_PREFIX_entry_location_rocksdb_TableOptions_ && ";
+        }
+
         final boolean tlsEnabledOnBookKeeper = isTlsEnabledOnBookKeeper();
         if (tlsEnabledOnBookKeeper) {
             mainArg +=