Merge pull request #517 from CloudSlang/oracle_memory_management

ORA-04031 fix
CloudSlang · Jan 20, 2025 · c6c8f80 · c6c8f80
2 parents bfbab09 + a0ade43
commit c6c8f80
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 11 deletions.
diff --git a/...-node-impl/src/main/java/io/cloudslang/engine/node/repositories/WorkerNodeRepository.java b/...-node-impl/src/main/java/io/cloudslang/engine/node/repositories/WorkerNodeRepository.java
@@ -21,6 +21,8 @@
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 
 import java.util.List;
 
@@ -56,4 +58,8 @@ public interface WorkerNodeRepository extends JpaRepository<WorkerNode,Long> {
 
 	@Modifying @Query("update WorkerNode w set w.uuid = w.uuid where w.uuid = ?1")
 	void lockByUuid(String uuid);
+
+    @Query(value = "SELECT w FROM WorkerNode w",
+            countQuery = "SELECT count(w) FROM WorkerNode w")
+    Page<WorkerNode> findAllWithPagination(Pageable pageable);
 }
diff --git a/...ore-node-impl/src/main/java/io/cloudslang/engine/node/services/WorkerNodeServiceImpl.java b/...ore-node-impl/src/main/java/io/cloudslang/engine/node/services/WorkerNodeServiceImpl.java
@@ -28,6 +28,8 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -52,6 +54,7 @@ public class WorkerNodeServiceImpl implements WorkerNodeService {
     private static final long MAX_VERSION_GAP_ALLOWED = Long.getLong("max.allowed.version.gap.worker.recovery", 2);
     private static boolean disableMonitoring = Boolean.getBoolean("global.worker.monitoring.disable");
     private static final String MSG_RECOVERY_VERSION_NAME = "MSG_RECOVERY_VERSION";
+    private static final int WORKER_GROUPS_PAGE_SIZE = Integer.getInteger("worker.groups.page.size", 50);
 
     @Autowired
     private WorkerNodeRepository workerNodeRepository;
@@ -271,7 +274,24 @@ public List<WorkerNode> readAllWorkers() {
     @Override
     @Transactional(readOnly = true)
     public Map<String, Set<String>> readWorkerGroupsMap() {
-        List<WorkerNode> all = workerNodeRepository.findAll();
+        /**
+         * Reduces memory usage by loading data in chunks of WORKER_GROUPS_PAGE_SIZE
+         * Prevent ORA-04031.
+         */
+        int currentPage = 0;
+        List<WorkerNode> all = new ArrayList<>();
+
+        Page<WorkerNode> page;
+        do {
+            page = workerNodeRepository.findAllWithPagination(
+                    PageRequest.of(currentPage, WORKER_GROUPS_PAGE_SIZE)
+            );
+            if (page != null && page.hasContent()) {
+                all.addAll(page.getContent());
+            }
+            currentPage++;
+        } while (page.hasNext());
+
         Map<String, Set<String>> workerGroupsMap = newHashMapWithExpectedSize(all.size());
         for (WorkerNode workerNode : all) {
             workerGroupsMap.put(workerNode.getUuid(), new HashSet<>(workerNode.getGroups()));
@@ -507,5 +527,4 @@ public void updateWRV(String workerUuid, String wrv) {
         WorkerNode worker = workerNodeRepository.findByUuid(workerUuid);
         worker.setWorkerRecoveryVersion(wrv);
     }
-
 }
diff --git a/...l/src/main/java/io/cloudslang/engine/queue/repositories/ExecutionQueueRepositoryImpl.java b/...l/src/main/java/io/cloudslang/engine/queue/repositories/ExecutionQueueRepositoryImpl.java
@@ -52,6 +52,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.Collections;
 
 import static java.lang.Long.parseLong;
 import static org.apache.commons.io.IOUtils.toByteArray;
@@ -334,8 +335,7 @@ public class ExecutionQueueRepositoryImpl implements ExecutionQueueRepository {
     final private String BUSY_WORKERS_SQL =
             "SELECT ASSIGNED_WORKER      " +
                     " FROM  OO_EXECUTION_QUEUES q  " +
-                    " WHERE  " +
-                    "      (q.STATUS IN (:status)) AND " +
+                    " WHERE q.STATUS IN (%s) AND " +
                     " (NOT EXISTS (SELECT qq.MSG_SEQ_ID " +
                     "              FROM OO_EXECUTION_QUEUES qq " +
                     "              WHERE (qq.EXEC_STATE_ID = q.EXEC_STATE_ID) AND qq.MSG_SEQ_ID > q.MSG_SEQ_ID)) " +
@@ -897,15 +897,19 @@ public List<ExecutionMessage> findByStatuses(int maxSize, ExecStatus... statuses
 
     @Override
     public List<String> getBusyWorkers(ExecStatus... statuses) {
+        /**
+         * Uses bind parameters to avoid memory issues. This way the database uses fewer unique statements and thus reduces memory usage
+         * This way the database can reuse the same statement for multiple executions and thus prevent ORA-04031.
+         */
+        String bindParams = String.join(",", Collections.nCopies(statuses.length, "?"));
         // prepare the sql statement
-        String sqlStat = BUSY_WORKERS_SQL
-                .replaceAll(":status", StringUtils.repeat("?", ",", statuses.length));
+        String sqlStat = String.format(BUSY_WORKERS_SQL, bindParams);
+
         // prepare the argument
-        Object[] values = new Object[statuses.length];
-        int i = 0;
-        for (ExecStatus status : statuses) {
-            values[i] = status.getNumber();
-        }
+        Object[] values = Arrays.stream(statuses)
+                .map(ExecStatus::getNumber)
+                .toArray();
+
         return doSelectWithTemplate(getBusyWorkersJdbcTemplate, sqlStat, new BusyWorkerRowMapper(), values);
     }