3258 - Fix OutOfMemoryError in the workflow editor when iterating with loop over large number of items

Author: ivicac

server/ee/apps/runtime-job-app/build.gradle.kts

@@ -36,7 +36,7 @@ dependencies {
     implementation(project(":server:libs:core:evaluator:evaluator-impl"))
     implementation(project(":server:libs:core:file-storage:file-storage-base64-service"))
     implementation(project(":server:libs:core:file-storage:file-storage-filesystem-service"))
-    implementation(project(":server:libs:core:message:message-broker:message-broker-sync"))
+    implementation(project(":server:libs:core:message:message-broker:message-broker-memory"))
     implementation(project(":server:libs:core:message:message-event:message-event-impl"))
     implementation(project(":server:libs:platform:platform-component:platform-component-api"))
     implementation(project(":server:libs:platform:platform-component:platform-component-context:platform-component-context-service"))

server/ee/apps/runtime-job-app/src/main/java/com/bytechef/runtime/job/platform/connection/ConnectionContext.java

@@ -7,8 +7,8 @@
 
 package com.bytechef.runtime.job.platform.connection;
 
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -19,22 +19,16 @@
 public class ConnectionContext {
 
     private static final AtomicLong ATOMIC_CONNECTION_ID = new AtomicLong(1);
-
-    private static final ThreadLocal<Map<Long, Connection>> CURRENT_CONNECTION_MAP = ThreadLocal.withInitial(
-        HashMap::new);
+    private static final Map<Long, Connection> CONNECTION_MAP = new ConcurrentHashMap<>();
 
     public static Map<String, ?> getConnectionParameters(long id) {
-        Map<Long, Connection> parameterMap = CURRENT_CONNECTION_MAP.get();
-
-        return parameterMap.get(id).parameters;
+        return CONNECTION_MAP.get(id).parameters;
     }
 
     public static long putConnectionParameters(String name, Map<String, ?> parameters) {
-        Map<Long, Connection> parameterMap = CURRENT_CONNECTION_MAP.get();
-
         long connectionId = -1;
 
-        for (Map.Entry<Long, Connection> entry : parameterMap.entrySet()) {
+        for (Map.Entry<Long, Connection> entry : CONNECTION_MAP.entrySet()) {
             Connection connection = entry.getValue();
 
             if (connection.name.equals(name)) {
@@ -47,9 +41,7 @@ public static long putConnectionParameters(String name, Map<String, ?> parameter
         if (connectionId == -1) {
             connectionId = ATOMIC_CONNECTION_ID.getAndIncrement();
 
-            parameterMap.putIfAbsent(connectionId, new Connection(name, parameters));
-
-            CURRENT_CONNECTION_MAP.set(parameterMap);
+            CONNECTION_MAP.putIfAbsent(connectionId, new Connection(name, parameters));
         }
 
         return connectionId;

server/ee/libs/embedded/embedded-configuration/embedded-configuration-service/build.gradle.kts

@@ -24,6 +24,7 @@ dependencies {
     testImplementation(project(":server:libs:core:evaluator:evaluator-impl"))
     testImplementation(project(":server:libs:platform:platform-category:platform-category-service"))
     testImplementation(project(":server:libs:platform:platform-configuration:platform-configuration-service"))
+    testImplementation(project(":server:libs:platform:platform-security:platform-security-service"))
     testImplementation(project(":server:libs:platform:platform-tag:platform-tag-service"))
     testImplementation(project(":server:libs:test:test-int-support"))
 }

server/libs/atlas/atlas-coordinator/atlas-coordinator-impl/build.gradle.kts

@@ -20,7 +20,7 @@ dependencies {
     testImplementation(project(":server:libs:atlas:atlas-execution:atlas-execution-service"))
     testImplementation(project(":server:libs:atlas:atlas-file-storage:atlas-file-storage-impl"))
     testImplementation(project(":server:libs:atlas:atlas-worker:atlas-worker-impl"))
-    testImplementation(project(":server:libs:core:message:message-broker:message-broker-sync"))
+    testImplementation(project(":server:libs:core:message:message-broker:message-broker-memory"))
     testImplementation(project(":server:libs:core:commons:commons-data"))
     testImplementation(project(":server:libs:core:evaluator:evaluator-impl"))
     testImplementation(project(":server:libs:core:file-storage:file-storage-base64-service"))

server/libs/atlas/atlas-coordinator/atlas-coordinator-impl/src/test/java/com/bytechef/atlas/coordinator/TaskCoordinatorIntTest.java

@@ -52,6 +52,7 @@
 import com.bytechef.file.storage.base64.service.Base64FileStorageService;
 import com.bytechef.jackson.config.JacksonConfiguration;
 import com.bytechef.liquibase.config.LiquibaseConfiguration;
+import com.bytechef.message.broker.memory.SyncMessageBroker;
 import com.bytechef.platform.coordinator.job.JobSyncExecutor;
 import com.bytechef.test.config.jdbc.AbstractIntTestJdbcConfiguration;
 import com.bytechef.test.config.testcontainers.PostgreSQLContainerConfiguration;
@@ -62,7 +63,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -75,7 +75,6 @@
 import org.springframework.core.env.Environment;
 import org.springframework.core.io.support.ResourcePatternResolver;
 import org.springframework.data.jdbc.repository.config.EnableJdbcRepositories;
-import org.springframework.lang.Nullable;
 
 /**
  * @author Arik Cohen
@@ -95,23 +94,18 @@ public class TaskCoordinatorIntTest {
     private final Evaluator evaluator = SpelEvaluator.create();
 
     @Autowired
-    @Nullable
     private Environment environment;
 
     @Autowired
-    @Nullable
     private ContextService contextService;
 
     @Autowired
-    @Nullable
     private JobService jobService;
 
     @Autowired
-    @Nullable
     private TaskExecutionService taskExecutionService;
 
     @Autowired
-    @Nullable
     private WorkflowService workflowService;
 
     @Test
@@ -134,9 +128,8 @@ private Job executeWorkflow(String workflowId) {
         taskHandlerMap.put("randomHelper/v1/randomInt", taskExecution -> null);
 
         JobSyncExecutor jobSyncExecutor = new JobSyncExecutor(
-            Objects.requireNonNull(contextService), environment, evaluator, Objects.requireNonNull(jobService),
-            List.of(), Objects.requireNonNull(taskExecutionService), taskHandlerMap::get, TASK_FILE_STORAGE,
-            Objects.requireNonNull(workflowService));
+            contextService, environment, evaluator, jobService, -1, SyncMessageBroker::new, List.of(),
+            taskExecutionService, taskHandlerMap::get, TASK_FILE_STORAGE, -1, workflowService);
 
         return jobSyncExecutor.execute(new JobParametersDTO(workflowId, Collections.singletonMap("yourName", "me")));
     }

server/libs/atlas/atlas-execution/atlas-execution-repository/atlas-execution-repository-memory/src/main/java/com/bytechef/atlas/execution/repository/memory/InMemoryJobRepository.java

@@ -71,7 +71,9 @@ public int countRunningJobs() {
 
     @Override
     public void deleteById(Long id) {
-        throw new UnsupportedOperationException();
+        Cache cache = Objects.requireNonNull(cacheManager.getCache(CACHE));
+
+        cache.evict(TenantCacheKeyUtils.getKey(id));
     }
 
     @Override

server/libs/atlas/atlas-worker/atlas-worker-impl/build.gradle.kts

@@ -17,6 +17,6 @@ dependencies {
     testImplementation(project(":server:libs:atlas:atlas-file-storage:atlas-file-storage-impl"))
     testImplementation(project(":server:libs:core:evaluator:evaluator-impl"))
     testImplementation(project(":server:libs:core:file-storage:file-storage-base64-service"))
-    testImplementation(project(":server:libs:core:message:message-broker:message-broker-sync"))
+    testImplementation(project(":server:libs:core:message:message-broker:message-broker-memory"))
     testImplementation(project(":server:libs:test:test-support"))
 }

server/libs/core/commons/commons-util/src/main/java/com/bytechef/commons/util/CollectionUtils.java

@@ -214,7 +214,9 @@ public static <T, R> List<R> map(List<T> list, Function<? super T, R> mapper) {
         Validate.notNull(list, "'list' must not be null");
         Validate.notNull(mapper, "'mapper' must not be null");
 
-        return list.stream()
+        // Snapshot to avoid ConcurrentModificationException when the source list is modified concurrently
+        return new ArrayList<>(list)
+            .stream()
             .map(mapper)
             .toList();
     }
@@ -223,7 +225,9 @@ public static <R, T> List<R> map(Set<T> set, Function<? super T, R> mapper) {
         Validate.notNull(set, "'set' must not be null");
         Validate.notNull(mapper, "'mapper' must not be null");
 
-        return set.stream()
+        // Snapshot to avoid ConcurrentModificationException when the source set is modified concurrently
+        return new ArrayList<>(set)
+            .stream()
             .map(mapper)
             .toList();
     }

server/libs/core/message/message-broker/message-broker-sync/build.gradle.kts renamed to server/libs/core/message/message-broker/message-broker-memory/build.gradle.kts

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2025 ByteChef
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.bytechef.message.broker.memory;
+
+import com.bytechef.message.route.MessageRoute;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Abstract base implementation of the {@link MessageBroker} interface. This class provides a framework for managing
+ * receivers and routing messages to them based on specified routes. Concrete subclasses need to implement the
+ * {@code send} method to define the specific behavior for message delivery.
+ *
+ * @author Ivica Cardic
+ */
+public abstract class AbstractMessageBroker implements MemoryMessageBroker {
+
+    protected final Map<MessageRoute, List<SyncMessageBroker.Receiver>> receiverMap = new HashMap<>();
+
+    @Override
+    public void receive(MessageRoute messageRoute, SyncMessageBroker.Receiver receiver) {
+        List<SyncMessageBroker.Receiver> receivers = receiverMap.computeIfAbsent(messageRoute, k -> new ArrayList<>());
+
+        receivers.add(receiver);
+    }
+}