From b79178f79848f7875b3a596e20418e58dc357d37 Mon Sep 17 00:00:00 2001 From: Eduardo Flores Date: Sat, 18 Jul 2026 00:45:00 -0600 Subject: [PATCH] feat(plugins): consolidate lifecycle lock ownership --- .../plugins/PluginRegistryService.java | 42 +--- .../loader/DynamicJobLoaderService.java | 124 ++++++---- .../plugins/PluginRegistryServiceTest.java | 35 +-- .../loader/DynamicJobLoaderServiceTest.java | 218 ++++++++++++++++-- 4 files changed, 313 insertions(+), 106 deletions(-) diff --git a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java index 7613c51..4f8ef71 100644 --- a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java +++ b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java @@ -44,13 +44,7 @@ public class PluginRegistryService { /** Built during {@link PostConstruct} and dynamic registration; mutable. */ private final Map pluginsByJobName = new LinkedHashMap<>(); - /** - * Tracks classloader references for dynamically-registered plugins. Stored for - * Phase 4 cleanup (classloader close/reload). Key: job name, Value: classloader - * reference. - */ - private final Map classloaderRefs = new HashMap<>(); - + /** Creates a registry backed by the shared Spring Batch infrastructure and job definitions. */ public PluginRegistryService( List plugins, JobRegistry jobRegistry, @@ -175,15 +169,12 @@ public Set getRegisteredJobNames() { *

Thread-safe: acquires the {@code pluginsByJobName} monitor before * mutation so register-into-map + register-into-JobRegistry are atomic. * - * @param plugin the plugin instance to register (must have a unique - * job name) - * @param classLoaderRef a reference to the classloader that loaded the - * plugin (stored for Phase 4 cleanup) + * @param plugin the plugin instance to register (must have a unique job name) * @throws PluginRegistrationException if a plugin with the same job name * is already registered, or if * {@code configureJob()} fails */ - public void registerDynamicPlugin(BatchJobPlugin plugin, Object classLoaderRef) { + public void registerDynamicPlugin(BatchJobPlugin plugin) { String name = plugin.getJobName(); synchronized (pluginsByJobName) { @@ -223,8 +214,6 @@ public void registerDynamicPlugin(BatchJobPlugin plugin, Object classLoaderRef) } pluginsByJobName.put(name, plugin); - classloaderRefs.put(name, classLoaderRef); - log.info( "Dynamically registered plugin job '{}' from {}", name, @@ -236,12 +225,9 @@ public void registerDynamicPlugin(BatchJobPlugin plugin, Object classLoaderRef) * Unregisters a dynamically-registered plugin from the internal map and the * shared {@link JobRegistry}. * - *

Thread-safe: acquires the {@code pluginsByJobName} monitor. The - * classloader reference is retained in {@code classloaderRefs} for - * Phase 4 cleanup (graceful drain / close). - * - *

If the job is currently running, a warning is logged but execution is - * not interrupted — Phase 4 handles graceful drain. + *

Thread-safe: acquires the {@code pluginsByJobName} monitor. The map entry + * is removed only after {@link JobRegistry#unregister(String)} succeeds, so a + * registry failure preserves the plugin's in-memory registration. * * @param jobName the job name to unregister */ @@ -252,22 +238,10 @@ public void unregisterDynamicPlugin(String jobName) { return; } + jobRegistry.unregister(jobName); pluginsByJobName.remove(jobName); - try { - jobRegistry.unregister(jobName); - } catch (Exception e) { - log.warn( - "Failed to unregister job \"{}\" from JobRegistry: {}", - jobName, - e.getMessage()); - } - - classloaderRefs.remove(jobName); - - log.info( - "Unregistered plugin '{}', classloader reference removed", - jobName); + log.info("Unregistered plugin '{}'", jobName); } } 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 b1027e7..fda0971 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 @@ -20,6 +20,7 @@ import java.util.NoSuchElementException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.batch.core.repository.JobRepository; @@ -59,7 +60,7 @@ public class DynamicJobLoaderService { * 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<>(); + private final Map lifecycleLocks = new ConcurrentHashMap<>(); public DynamicJobLoaderService( JobDefinitionRepository jobDefinitionRepository, @@ -91,20 +92,48 @@ public DynamicJobLoaderService( */ 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 + // bogus id) cannot grow lifecycleLocks 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(); + return withDefinitionLock(definitionId, () -> doLoadJob(definitionId)); + } + + T withDefinitionLock(Long definitionId, Supplier action) { + LifecycleLockHolder holder = + lifecycleLocks.compute( + definitionId, + (id, current) -> { + LifecycleLockHolder selected = current == null ? new LifecycleLockHolder() : current; + selected.users++; + return selected; + }); + holder.lock.lock(); try { - return doLoadJob(definitionId); + return action.get(); } finally { - lock.unlock(); + holder.lock.unlock(); + lifecycleLocks.computeIfPresent( + definitionId, + (id, current) -> --current.users == 0 ? null : current); } } + void withDefinitionLock(Long definitionId, Runnable action) { + withDefinitionLock( + definitionId, + () -> { + action.run(); + return null; + }); + } + + private static final class LifecycleLockHolder { + private final ReentrantLock lock = new ReentrantLock(); + private int users; + } + /** * 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 @@ -151,6 +180,7 @@ private LoadResult doLoadJob(Long definitionId) { long loadStartNanos = System.nanoTime(); DynamicJobClassLoader classLoader = null; + boolean registered = false; try { // Validate JAR file exists File jarFile = new File(entity.getJarFilePath()); @@ -206,7 +236,8 @@ private LoadResult doLoadJob(Long definitionId) { // Register through existing hook — registerDynamicPlugin() internally calls // plugin.configureJob(...), validates the Job, and registers in JobRegistry - pluginRegistryService.registerDynamicPlugin(plugin, classLoader); + pluginRegistryService.registerDynamicPlugin(plugin); + registered = true; // Track classloader for future unload classLoaders.put(definitionId, classLoader); @@ -240,35 +271,52 @@ private LoadResult doLoadJob(Long definitionId) { return new LoadResult( entity.getJobName(), LoadResult.LOADED, "Successfully loaded"); - } catch (NoSuchElementException | IllegalStateException e) { - // Re-throw without wrapping — these are validation failures, not load errors - // Reset status from LOADING back to previous - entity.setLoadStatus(getPreviousLoadStatus(entity, definitionId)); - jobDefinitionRepository.save(entity); - auditService.logEvent( - new AuditEvent( - AuditEventType.JOB_LOADED, - entity.getJobName(), - AuditService.currentUserId(), - "Load rejected: " + e.getMessage(), - AuditEvent.FAILURE, - LocalDateTime.now())); - throw e; } catch (Exception e) { - // Mark FAILED and persist the error - entity.setLoadStatus(LoadResult.FAILED); - entity.setLoadError(e.getMessage()); - jobDefinitionRepository.save(entity); - - // Clean up any partial classloader - if (classLoader != null) { + boolean canReleaseClassLoader = true; + if (registered) { try { - classLoader.cleanup(); - } catch (Exception cleanupEx) { - log.warn("Failed to clean up classloader for definition {}: {}", definitionId, cleanupEx.getMessage()); + pluginRegistryService.unregisterDynamicPlugin(entity.getJobName()); + } catch (Exception rollbackEx) { + canReleaseClassLoader = false; + e.addSuppressed(rollbackEx); + log.error( + "Failed to roll back plugin registration for definition {}: {}", + definitionId, + rollbackEx.getMessage(), + rollbackEx); + } + } + + if (canReleaseClassLoader) { + DynamicJobClassLoader ownedClassLoader = classLoaders.remove(definitionId); + DynamicJobClassLoader classLoaderToClean = + ownedClassLoader != null ? ownedClassLoader : classLoader; + if (classLoaderToClean != null) { + try { + classLoaderToClean.cleanup(); + } catch (Exception cleanupEx) { + e.addSuppressed(cleanupEx); + log.warn( + "Failed to clean up classloader for definition {}: {}", + definitionId, + cleanupEx.getMessage(), + cleanupEx); + } } } - classLoaders.remove(definitionId); + + entity.setLoadStatus(LoadResult.FAILED); + entity.setLoadError(e.getMessage()); + try { + jobDefinitionRepository.save(entity); + } catch (Exception persistenceEx) { + e.addSuppressed(persistenceEx); + log.error( + "Failed to persist FAILED status for definition {}: {}", + definitionId, + persistenceEx.getMessage(), + persistenceEx); + } log.error("Failed to load plugin for definition {}: {}", definitionId, e.getMessage(), e); @@ -440,16 +488,4 @@ public Map loadAllEnabled() { return results; } - /** - * Determines the previous load status for rollback when a pre-condition validation fails after - * LOADING was already set. - */ - private String getPreviousLoadStatus(JobDefinitionEntity entity, Long definitionId) { - // If we already had a classloader, it means we were LOADED before this attempt - if (classLoaders.containsKey(definitionId)) { - return LoadResult.LOADED; - } - // Otherwise revert to UNLOADED (the most common prior state for a fresh load) - return LoadResult.UNLOADED; - } } diff --git a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java index 95d4226..895c4aa 100644 --- a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java +++ b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java @@ -9,6 +9,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -120,13 +121,12 @@ void configureJobThrows_wrapsInPluginRegistrationException_withPluginClassAndCau void registerDynamicPlugin_addsToMapAndJobRegistry() throws Exception { Job job = stubJob("dynamicJob"); BatchJobPlugin plugin = stubPlugin("dynamicJob", job); - Object classLoaderRef = new Object(); PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, applicationContext, jobDefinitionRepository); - service.registerDynamicPlugin(plugin, classLoaderRef); + service.registerDynamicPlugin(plugin); assertTrue(service.getRegisteredJobNames().contains("dynamicJob")); assertTrue(service.getPlugin("dynamicJob").isPresent()); @@ -134,19 +134,20 @@ void registerDynamicPlugin_addsToMapAndJobRegistry() throws Exception { } @Test - void registerDynamicPlugin_tracksClassloaderReference() throws Exception { + void registerDynamicPlugin_doesNotRetainClassloaderOwnership() throws Exception { Job job = stubJob("dynamicJob"); BatchJobPlugin plugin = stubPlugin("dynamicJob", job); - Object classLoaderRef = new Object(); PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, applicationContext, jobDefinitionRepository); - service.registerDynamicPlugin(plugin, classLoaderRef); + service.registerDynamicPlugin(plugin); - // getPluginBySource confirms it's registered as classpath assertEquals("CLASSPATH", service.getPluginBySource("dynamicJob")); + assertThrows( + NoSuchFieldException.class, + () -> PluginRegistryService.class.getDeclaredField("classloaderRefs")); } @Test @@ -160,11 +161,11 @@ void registerDynamicPlugin_throwsOnDuplicate() throws Exception { List.of(), jobRegistry, jobRepository, transactionManager, applicationContext, jobDefinitionRepository); - service.registerDynamicPlugin(plugin1, new Object()); + service.registerDynamicPlugin(plugin1); PluginRegistrationException ex = assertThrows( PluginRegistrationException.class, - () -> service.registerDynamicPlugin(plugin2, new Object())); + () -> service.registerDynamicPlugin(plugin2)); assertTrue(ex.getMessage().contains("dynamicJob"), "message should contain the conflicting job name"); @@ -179,7 +180,7 @@ void unregisterDynamicPlugin_removesFromMapAndJobRegistry() throws Exception { List.of(), jobRegistry, jobRepository, transactionManager, applicationContext, jobDefinitionRepository); - service.registerDynamicPlugin(plugin, new Object()); + service.registerDynamicPlugin(plugin); service.unregisterDynamicPlugin("dynamicJob"); assertFalse(service.getRegisteredJobNames().contains("dynamicJob")); @@ -188,20 +189,24 @@ void unregisterDynamicPlugin_removesFromMapAndJobRegistry() throws Exception { } @Test - void unregisterDynamicPlugin_retainsClassloaderReference() throws Exception { + void unregisterDynamicPlugin_propagatesRegistryFailureAndPreservesPluginMap() throws Exception { Job job = stubJob("dynamicJob"); BatchJobPlugin plugin = stubPlugin("dynamicJob", job); - Object classLoaderRef = new Object(); PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, applicationContext, jobDefinitionRepository); - service.registerDynamicPlugin(plugin, classLoaderRef); - service.unregisterDynamicPlugin("dynamicJob"); + service.registerDynamicPlugin(plugin); + RuntimeException failure = new RuntimeException("registry unavailable"); + doThrow(failure).when(jobRegistry).unregister("dynamicJob"); - // Plugin is unregistered from the map but the classloader ref is retained - assertFalse(service.getRegisteredJobNames().contains("dynamicJob")); + RuntimeException thrown = + assertThrows(RuntimeException.class, () -> service.unregisterDynamicPlugin("dynamicJob")); + + assertEquals(failure, thrown); + assertTrue(service.getRegisteredJobNames().contains("dynamicJob")); + assertTrue(service.getPlugin("dynamicJob").isPresent()); } @Test 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 d7b88f9..918ec8c 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 @@ -104,14 +104,22 @@ private JobDefinitionEntity createEntity( return entity; } - /** Reads the private per-definition {@code loadLocks} map via reflection for white-box assertions. */ + /** Reads the reference-counted lifecycle holder map for white-box assertions. */ @SuppressWarnings("unchecked") - private static Map loadLocksOf(DynamicJobLoaderService target) throws Exception { - var field = DynamicJobLoaderService.class.getDeclaredField("loadLocks"); + private static Map lifecycleLocksOf(DynamicJobLoaderService target) throws Exception { + var field = DynamicJobLoaderService.class.getDeclaredField("lifecycleLocks"); field.setAccessible(true); return (Map) field.get(target); } + @SuppressWarnings("unchecked") + private static Map classLoadersOf( + DynamicJobLoaderService target) throws Exception { + var field = DynamicJobLoaderService.class.getDeclaredField("classLoaders"); + field.setAccessible(true); + return (Map) field.get(target); + } + // ── Successful load ─────────────────────────────────────────────────────── @Test @@ -121,7 +129,7 @@ void loadJob_validPlugin_registersAndReturnsLoaded() throws Exception { when(jobDefinitionRepository.findById(1L)).thenReturn(Optional.of(entity)); doNothing() .when(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any()); + .registerDynamicPlugin(any(BatchJobPlugin.class)); try (MockedConstruction mocked = mockConstruction( @@ -140,7 +148,7 @@ void loadJob_validPlugin_registersAndReturnsLoaded() throws Exception { assertNull(entity.getLoadError()); verify(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any(DynamicJobClassLoader.class)); + .registerDynamicPlugin(any(BatchJobPlugin.class)); } } @@ -254,7 +262,7 @@ void loadJob_configureJobFails_setsFailed() throws Exception { when(jobDefinitionRepository.findById(17L)).thenReturn(Optional.of(entity)); doThrow(new PluginRegistrationException("configureJob failed")) .when(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any()); + .registerDynamicPlugin(any(BatchJobPlugin.class)); try (MockedConstruction mocked = mockConstruction( @@ -273,6 +281,73 @@ void loadJob_configureJobFails_setsFailed() throws Exception { } } + @Test + void loadJob_lateFailureAfterRegistration_unregistersBeforeClassLoaderCleanup() throws Exception { + JobDefinitionEntity entity = createEntity(19L, "test-job", true, null); + RuntimeException loadFailure = new RuntimeException("persisting LOADED failed"); + + when(jobDefinitionRepository.findById(19L)).thenReturn(Optional.of(entity)); + when(jobDefinitionRepository.save(entity)) + .thenAnswer( + invocation -> { + if (LoadResult.LOADED.equals(entity.getLoadStatus())) { + throw loadFailure; + } + return entity; + }); + + try (MockedConstruction mocked = + mockConstruction( + DynamicJobClassLoader.class, + (mock, ctx) -> + when(mock.loadClass("com.test.TestPlugin")) + .thenAnswer(inv -> TestBatchJobPlugin.class))) { + JobLoadException thrown = + assertThrows(JobLoadException.class, () -> service.loadJob(19L)); + DynamicJobClassLoader candidate = mocked.constructed().getFirst(); + InOrder rollbackOrder = inOrder(pluginRegistryService, candidate); + + assertSame(loadFailure, thrown.getCause()); + rollbackOrder.verify(pluginRegistryService).unregisterDynamicPlugin("test-job"); + rollbackOrder.verify(candidate).cleanup(); + assertFalse(classLoadersOf(service).containsKey(19L)); + } + } + + @Test + void loadJob_rollbackUnregisterFails_retainsClassLoaderOwnership() throws Exception { + JobDefinitionEntity entity = createEntity(20L, "test-job", true, null); + RuntimeException loadFailure = new RuntimeException("persisting LOADED failed"); + RuntimeException rollbackFailure = new RuntimeException("unregister failed"); + + when(jobDefinitionRepository.findById(20L)).thenReturn(Optional.of(entity)); + when(jobDefinitionRepository.save(entity)) + .thenAnswer( + invocation -> { + if (LoadResult.LOADED.equals(entity.getLoadStatus())) { + throw loadFailure; + } + return entity; + }); + doThrow(rollbackFailure).when(pluginRegistryService).unregisterDynamicPlugin("test-job"); + + try (MockedConstruction mocked = + mockConstruction( + DynamicJobClassLoader.class, + (mock, ctx) -> + when(mock.loadClass("com.test.TestPlugin")) + .thenAnswer(inv -> TestBatchJobPlugin.class))) { + JobLoadException thrown = + assertThrows(JobLoadException.class, () -> service.loadJob(20L)); + DynamicJobClassLoader candidate = mocked.constructed().getFirst(); + + assertSame(loadFailure, thrown.getCause()); + assertSame(rollbackFailure, loadFailure.getSuppressed()[0]); + assertSame(candidate, classLoadersOf(service).get(20L)); + verify(candidate, never()).cleanup(); + } + } + // ── Pre-condition validations ───────────────────────────────────────────── @Test @@ -289,7 +364,7 @@ void loadJob_nonexistentId_doesNotRetainLock() throws Exception { assertThrows(NoSuchElementException.class, () -> service.loadJob(999L)); assertTrue( - loadLocksOf(service).isEmpty(), + lifecycleLocksOf(service).isEmpty(), "loadJob must not retain a per-definition lock for a nonexistent id"); } @@ -386,7 +461,7 @@ void reloadJob_unloadsThenLoads_atomic() throws Exception { doNothing().when(pluginRegistryService).unregisterDynamicPlugin("test-job"); doNothing() .when(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any()); + .registerDynamicPlugin(any(BatchJobPlugin.class)); try (MockedConstruction mocked = mockConstruction( @@ -401,7 +476,7 @@ void reloadJob_unloadsThenLoads_atomic() throws Exception { assertEquals(LoadResult.LOADED, result.status()); verify(pluginRegistryService).unregisterDynamicPlugin("test-job"); verify(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any(DynamicJobClassLoader.class)); + .registerDynamicPlugin(any(BatchJobPlugin.class)); } } @@ -420,7 +495,7 @@ void loadAllEnabled_skipsAlreadyLoaded_loadsOthers() throws Exception { when(pluginRegistryService.getRegisteredJobNames()).thenReturn(Set.of("already-loaded")); doNothing() .when(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any()); + .registerDynamicPlugin(any(BatchJobPlugin.class)); try (MockedConstruction mocked = mockConstruction( @@ -453,7 +528,7 @@ void loadAllEnabled_oneFails_othersStillLoaded() throws Exception { when(jobDefinitionRepository.findById(15L)).thenReturn(Optional.of(bad)); doNothing() .when(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any()); + .registerDynamicPlugin(any(BatchJobPlugin.class)); try (MockedConstruction mocked = mockConstruction( @@ -476,6 +551,123 @@ void loadAllEnabled_oneFails_othersStillLoaded() throws Exception { // ── Concurrency ─────────────────────────────────────────────────────────── + @Test + void withDefinitionLock_sameIdExcludesConcurrentCallers() throws Exception { + CountDownLatch firstEntered = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + CountDownLatch secondEntered = new CountDownLatch(1); + ExecutorService pool = Executors.newFixedThreadPool(2); + try { + Future first = + pool.submit( + () -> + service.withDefinitionLock( + 30L, + () -> { + firstEntered.countDown(); + await(releaseFirst); + })); + assertTrue(firstEntered.await(5, TimeUnit.SECONDS)); + Future second = pool.submit(() -> service.withDefinitionLock(30L, secondEntered::countDown)); + + assertFalse(secondEntered.await(200, TimeUnit.MILLISECONDS)); + releaseFirst.countDown(); + first.get(5, TimeUnit.SECONDS); + second.get(5, TimeUnit.SECONDS); + assertEquals(0, lifecycleLocksOf(service).size()); + } finally { + releaseFirst.countDown(); + pool.shutdownNow(); + } + } + + @Test + void withDefinitionLock_differentIdsProceedIndependently() throws Exception { + CountDownLatch bothEntered = new CountDownLatch(2); + CountDownLatch releaseBoth = new CountDownLatch(1); + ExecutorService pool = Executors.newFixedThreadPool(2); + try { + Future first = + pool.submit( + () -> + service.withDefinitionLock( + 31L, + () -> { + bothEntered.countDown(); + await(releaseBoth); + })); + Future second = + pool.submit( + () -> + service.withDefinitionLock( + 32L, + () -> { + bothEntered.countDown(); + await(releaseBoth); + })); + + assertTrue(bothEntered.await(5, TimeUnit.SECONDS)); + releaseBoth.countDown(); + first.get(5, TimeUnit.SECONDS); + second.get(5, TimeUnit.SECONDS); + assertTrue(lifecycleLocksOf(service).isEmpty()); + } finally { + releaseBoth.countDown(); + pool.shutdownNow(); + } + } + + @Test + void withDefinitionLock_queuedCallersShareHolderUntilAllComplete() throws Exception { + CountDownLatch firstEntered = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + CountDownLatch queuedEntered = new CountDownLatch(2); + ExecutorService pool = Executors.newFixedThreadPool(3); + try { + Future first = + pool.submit( + () -> + service.withDefinitionLock( + 33L, + () -> { + firstEntered.countDown(); + await(releaseFirst); + })); + assertTrue(firstEntered.await(5, TimeUnit.SECONDS)); + Future second = pool.submit(() -> service.withDefinitionLock(33L, queuedEntered::countDown)); + Future third = pool.submit(() -> service.withDefinitionLock(33L, queuedEntered::countDown)); + + assertEquals(1, lifecycleLocksOf(service).size()); + releaseFirst.countDown(); + assertTrue(queuedEntered.await(5, TimeUnit.SECONDS)); + first.get(5, TimeUnit.SECONDS); + second.get(5, TimeUnit.SECONDS); + third.get(5, TimeUnit.SECONDS); + assertTrue(lifecycleLocksOf(service).isEmpty()); + } finally { + releaseFirst.countDown(); + pool.shutdownNow(); + } + } + + @Test + void withDefinitionLock_reclaimsHolderAfterFailure() throws Exception { + assertThrows( + IllegalStateException.class, + () -> service.withDefinitionLock(34L, () -> { throw new IllegalStateException("failure"); })); + + assertTrue(lifecycleLocksOf(service).isEmpty()); + } + + private static void await(CountDownLatch latch) { + try { + latch.await(); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new AssertionError("Interrupted while awaiting test synchronization", exception); + } + } + /** * 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 @@ -523,7 +715,7 @@ void loadJob_concurrentSameId_serializesAndOpensSingleClassloader() throws Excep return null; }) .when(pluginRegistryService) - .registerDynamicPlugin(any(BatchJobPlugin.class), any()); + .registerDynamicPlugin(any(BatchJobPlugin.class)); ExecutorService pool = Executors.newFixedThreadPool(2); try { @@ -557,7 +749,7 @@ void loadJob_concurrentSameId_serializesAndOpensSingleClassloader() throws Excep + "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()); + .registerDynamicPlugin(any(BatchJobPlugin.class)); // Release the real classloader/temp-JAR handle opened by the successful load. service.unloadJob(20L, true);