fix(plugins): serialize loadJob per definition id to prevent file-handle leak

fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java

@@ -19,6 +19,7 @@
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.batch.core.repository.JobRepository;
@@ -50,6 +51,16 @@ public class DynamicJobLoaderService {
   /** Tracks active classloaders by definition ID for cleanup on unload. */
   private final Map<Long, DynamicJobClassLoader> classLoaders = new ConcurrentHashMap<>();
 
+  /**
+   * Per-definition load locks. Serializes concurrent {@link #loadJob(Long)} calls for the same
+   * definition ID so the already-loaded guard and the classloader/registry writes execute
+   * atomically. Without this, two concurrent loads of the same definition both pass the guard, each
+   * opens a JAR file-handle, and the loser's {@code classLoaders.put} silently overwrites the
+   * winner's entry — leaking the winner's handle. {@link #loadJob(Long)} allocates an entry only
+   * after confirming the definition exists, so unknown IDs cannot grow the map without bound.
+   */
+  private final Map<Long, ReentrantLock> loadLocks = new ConcurrentHashMap<>();
+
   public DynamicJobLoaderService(
       JobDefinitionRepository jobDefinitionRepository,
       PluginRegistryService pluginRegistryService,
@@ -79,6 +90,27 @@ public DynamicJobLoaderService(
    *     failure, etc.)
    */
   public LoadResult loadJob(Long definitionId) {
+    // Gate lock allocation on existence so unknown IDs (e.g. POST /definitions/{id}/load with a
+    // bogus id) cannot grow loadLocks without bound. The authoritative existence check still runs
+    // inside doLoadJob under the lock, guarding against a definition deleted between the two reads.
+    if (jobDefinitionRepository.findById(definitionId).isEmpty()) {
+      throw new NoSuchElementException("Job definition not found for id: " + definitionId);
+    }
+    ReentrantLock lock = loadLocks.computeIfAbsent(definitionId, k -> new ReentrantLock());
+    lock.lock();
+    try {
+      return doLoadJob(definitionId);
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  /**
+   * Performs the actual load. Always invoked while holding the per-definition lock acquired in
+   * {@link #loadJob(Long)}, so the already-loaded guard and the classloader/registry writes are
+   * atomic with respect to other loads of the same definition.
+   */
+  private LoadResult doLoadJob(Long definitionId) {
     JobDefinitionEntity entity =
         jobDefinitionRepository
             .findById(definitionId)

fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java

@@ -19,6 +19,13 @@
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.jupiter.api.*;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.*;
@@ -97,6 +104,14 @@ private JobDefinitionEntity createEntity(
     return entity;
   }
 
+  /** Reads the private per-definition {@code loadLocks} map via reflection for white-box assertions. */
+  @SuppressWarnings("unchecked")
+  private static Map<Long, ?> loadLocksOf(DynamicJobLoaderService target) throws Exception {
+    var field = DynamicJobLoaderService.class.getDeclaredField("loadLocks");
+    field.setAccessible(true);
+    return (Map<Long, ?>) field.get(target);
+  }
+
   // ── Successful load ───────────────────────────────────────────────────────
 
   @Test
@@ -267,6 +282,17 @@ void loadJob_definitionNotFound_throwsNoSuchElementException() {
     assertThrows(NoSuchElementException.class, () -> service.loadJob(999L));
   }
 
+  @Test
+  void loadJob_nonexistentId_doesNotRetainLock() throws Exception {
+    when(jobDefinitionRepository.findById(999L)).thenReturn(Optional.empty());
+
+    assertThrows(NoSuchElementException.class, () -> service.loadJob(999L));
+
+    assertTrue(
+        loadLocksOf(service).isEmpty(),
+        "loadJob must not retain a per-definition lock for a nonexistent id");
+  }
+
   @Test
   void loadJob_disabledDefinition_throwsIllegalStateException() {
     JobDefinitionEntity entity = createEntity(6L, "disabled-job", false, null);
@@ -448,6 +474,100 @@ void loadAllEnabled_oneFails_othersStillLoaded() throws Exception {
     }
   }
 
+  // ── Concurrency ───────────────────────────────────────────────────────────
+
+  /**
+   * Two concurrent {@code loadJob} calls for the SAME definition id must be serialized: only one may
+   * proceed through the load body (opening a classloader / JAR file-handle and registering); the
+   * loser must observe the winner's registration and be rejected, never opening a second handle.
+   *
+   * <p>Uses a REAL {@link DynamicJobClassLoader} rather than {@code mockConstruction}, because
+   * Mockito's construction mocking is thread-confined and would not intercept the pool threads. The
+   * parent-last classloader resolves {@link TestBatchJobPlugin} from the application classloader, so
+   * the load succeeds on both threads. The winner is parked inside {@code registerDynamicPlugin}
+   * (holding the per-id lock under the fix); the registration only becomes visible once it returns,
+   * modelling the real race window. Without serialization the contender sails past the already-loaded
+   * guard and reaches {@code registerDynamicPlugin} a second time; serialized, it parks on the lock
+   * and is rejected by the guard. The bounded {@link CountDownLatch} awaits assert this
+   * deterministically, without timing-dependent sleeps.
+   */
+  @Test
+  void loadJob_concurrentSameId_serializesAndOpensSingleClassloader() throws Exception {
+    JobDefinitionEntity entity = createEntity(20L, "test-job", true, null);
+    // Real parent-last classloader resolves this class from the application classloader.
+    entity.setMainClassName(TestBatchJobPlugin.class.getName());
+    when(jobDefinitionRepository.findById(20L)).thenReturn(Optional.of(entity));
+
+    // Stateful registry: a registration only becomes visible once register() returns, modelling the
+    // real race window between the already-loaded guard and the registry write.
+    Set<String> registeredNames = ConcurrentHashMap.newKeySet();
+    AtomicInteger guardCalls = new AtomicInteger();
+    CountDownLatch secondGuardReached = new CountDownLatch(1);
+    when(pluginRegistryService.getRegisteredJobNames())
+        .thenAnswer(
+            inv -> {
+              if (guardCalls.incrementAndGet() == 2) {
+                secondGuardReached.countDown();
+              }
+              return new HashSet<>(registeredNames);
+            });
+
+    CountDownLatch registerEntered = new CountDownLatch(1);
+    CountDownLatch releaseRegister = new CountDownLatch(1);
+    doAnswer(
+            inv -> {
+              registerEntered.countDown();
+              releaseRegister.await();
+              BatchJobPlugin plugin = inv.getArgument(0);
+              registeredNames.add(plugin.getJobName());
+              return null;
+            })
+        .when(pluginRegistryService)
+        .registerDynamicPlugin(any(BatchJobPlugin.class), any());
+
+    ExecutorService pool = Executors.newFixedThreadPool(2);
+    try {
+      // Winner: blocks inside registerDynamicPlugin while holding the per-id lock (under the fix).
+      Future<?> winner = pool.submit(() -> service.loadJob(20L));
+      assertTrue(
+          registerEntered.await(5, TimeUnit.SECONDS),
+          "Winning thread never reached registerDynamicPlugin");
+
+      // Contender: same id. Serialized, it parks on the lock and never reaches the guard again;
+      // unserialized, it passes the guard and reaches registerDynamicPlugin a second time.
+      Future<?> contender =
+          pool.submit(
+              () -> {
+                try {
+                  service.loadJob(20L);
+                } catch (IllegalStateException alreadyLoaded) {
+                  // Expected once serialized: the contender sees the winner's registration.
+                }
+              });
+
+      boolean contenderPassedGuard = secondGuardReached.await(1, TimeUnit.SECONDS);
+
+      releaseRegister.countDown();
+      winner.get(5, TimeUnit.SECONDS);
+      contender.get(5, TimeUnit.SECONDS);
+
+      assertFalse(
+          contenderPassedGuard,
+          "Second concurrent load entered the load body while the first held the lock — "
+              + "loadJob is not serialized per definition id");
+      // Serialized: only the winner registers; the contender is rejected by the guard.
+      verify(pluginRegistryService, times(1))
+          .registerDynamicPlugin(any(BatchJobPlugin.class), any());
+
+      // Release the real classloader/temp-JAR handle opened by the successful load.
+      service.unloadJob(20L, true);
+    } finally {
+      // Unblock any thread still parked in registerDynamicPlugin so shutdown cannot hang.
+      releaseRegister.countDown();
+      pool.shutdownNow();
+    }
+  }
+
   // ── Inner test plugin ─────────────────────────────────────────────────────
 
   /**