From 2d7029ecd78259207ea0ff44e40a9bd3ebc79535 Mon Sep 17 00:00:00 2001 From: Eduardo Flores Date: Thu, 2 Jul 2026 01:07:27 -0600 Subject: [PATCH 1/2] fix(plugins): serialize loadJob per definition id to prevent file-handle leak MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two concurrent loadJob calls for the same definition id could both pass the already-loaded guard, each open a DynamicJobClassLoader (JAR file-handle), and the losing thread's classLoaders.put would overwrite the winner's entry without closing it — leaking the winner's handle. Wrap the load body (extracted into doLoadJob) in a per-definition ReentrantLock so the guard, classloader creation, registry write, and classLoaders.put execute atomically for a given id. The loser now observes the winner's registration and is rejected before opening a second handle. Adds a deterministic concurrency test that reproduces the leak by parking the winner inside registerDynamicPlugin and asserting the contender cannot enter the load body. --- .../loader/DynamicJobLoaderService.java | 26 ++++++ .../loader/DynamicJobLoaderServiceTest.java | 91 +++++++++++++++++++ 2 files changed, 117 insertions(+) diff --git a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java index 3dec084..93d23cf 100644 --- a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java +++ b/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 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. Definitions are admin-managed and bounded, so the + * map does not grow unbounded in practice. + */ + private final Map loadLocks = new ConcurrentHashMap<>(); + public DynamicJobLoaderService( JobDefinitionRepository jobDefinitionRepository, PluginRegistryService pluginRegistryService, @@ -79,6 +90,21 @@ public DynamicJobLoaderService( * failure, etc.) */ public LoadResult loadJob(Long 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) diff --git a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java index 2c50832..ed13a28 100644 --- a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java +++ b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java @@ -19,6 +19,8 @@ import java.nio.file.Path; import java.time.Duration; import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.*; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.*; @@ -448,6 +450,95 @@ 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. + * + *

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 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()); + } finally { + pool.shutdownNow(); + } + } + // ── Inner test plugin ───────────────────────────────────────────────────── /** From 7b5c3013a2242d739f41639e8f107f7f59cafe62 Mon Sep 17 00:00:00 2001 From: Eduardo Flores Date: Fri, 3 Jul 2026 22:37:48 -0600 Subject: [PATCH 2/2] fix(plugins): bound loadLocks map to existing definition ids Gate per-definition lock allocation on findById existence so that a load request for an unknown id (e.g. POST /definitions/{id}/load with a bogus id) no longer stores a ReentrantLock in loadLocks forever. The authoritative existence check still runs inside doLoadJob under the lock. Also clean up the concurrent-load test: unload after the assertions to release the real classloader/temp-JAR handle, release the register latch in finally so shutdown cannot hang, and replace the wildcard java.util.concurrent import with explicit imports. --- .../loader/DynamicJobLoaderService.java | 10 ++++-- .../loader/DynamicJobLoaderServiceTest.java | 31 ++++++++++++++++++- 2 files changed, 38 insertions(+), 3 deletions(-) diff --git a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java index 93d23cf..b1027e7 100644 --- a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java +++ b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderService.java @@ -56,8 +56,8 @@ public class DynamicJobLoaderService { * 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. Definitions are admin-managed and bounded, so the - * map does not grow unbounded in practice. + * 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 loadLocks = new ConcurrentHashMap<>(); @@ -90,6 +90,12 @@ 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 { diff --git a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java index ed13a28..d7b88f9 100644 --- a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java +++ b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoaderServiceTest.java @@ -19,7 +19,12 @@ import java.nio.file.Path; import java.time.Duration; import java.util.*; -import java.util.concurrent.*; +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; @@ -99,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 loadLocksOf(DynamicJobLoaderService target) throws Exception { + var field = DynamicJobLoaderService.class.getDeclaredField("loadLocks"); + field.setAccessible(true); + return (Map) field.get(target); + } + // ── Successful load ─────────────────────────────────────────────────────── @Test @@ -269,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); @@ -534,7 +558,12 @@ void loadJob_concurrentSameId_serializesAndOpensSingleClassloader() throws Excep // 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(); } }