This is an automated email from the ASF dual-hosted git repository. sergehuber pushed a commit to branch UNOMI-1028-bundle-listener-race in repository https://gitbox.apache.org/repos/asf/unomi.git
commit d9ebed1039d544aee1790832412325baeeb34320 Author: Serge Huber <[email protected]> AuthorDate: Thu Oct 29 11:37:12 2026 +0100 UNOMI-1028: Load built-in types from bundles that start during the startup scan AbstractMultiTypeCachingService.postConstruct scanned the started bundles and only then registered its bundle listener. When the services Blueprint container waits for the persistence service, postConstruct runs on a Blueprint extender thread while the Karaf feature thread keeps starting bundles; a bundle such as plugins-base that started inside that window was neither scanned nor notified, so its condition and action types were never loaded ("ConditionType not found: nestedCondition" across the ITs). Register the listener before the scan and remember which bundles are loaded, so a bundle seen by both the scan and the listener loads once. The mark is cleared when the bundle stops and when the service is re-initialised. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../cache/AbstractMultiTypeCachingService.java | 26 +++-- .../cache/AbstractMultiTypeCachingServiceTest.java | 112 +++++++++++++++++++++ 2 files changed, 131 insertions(+), 7 deletions(-) diff --git a/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java b/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java index 3e2e88add..a95653ce3 100644 --- a/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java +++ b/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java @@ -103,6 +103,12 @@ public abstract class AbstractMultiTypeCachingService extends AbstractContextAwa */ protected final Map<Long, List<PluginType>> pluginTypes = new ConcurrentHashMap<>(); + /** + * IDs of the bundles whose predefined items are loaded. A bundle that starts during the startup + * scan is seen both by the scan and by the bundle listener; this set makes the second one a no-op. + */ + private final Set<Long> loadedBundleIds = ConcurrentHashMap.newKeySet(); + /** * Map tracking scheduled tasks for cache refreshes. * Key is the task name, value is the ScheduledTask instance. @@ -163,17 +169,21 @@ public abstract class AbstractMultiTypeCachingService extends AbstractContextAwa } /** - * Initializes caches, loads predefined items from bundles, registers a bundle listener, - * loads initial persistence data, and starts refresh timers. + * Initializes caches, registers a bundle listener, loads predefined items from the bundles + * already started, loads initial persistence data, and starts refresh timers. */ public void postConstruct() { logger.debug("postConstruct {{}}", bundleContext.getBundle()); - // Initialize caches and load predefined items initializeCaches(); - loadPredefinedItems(bundleContext); + loadedBundleIds.clear(); - // Process existing bundles + // Listen before scanning: a bundle that starts while the scan runs (other bundles keep + // starting on the Karaf feature thread) is then either scanned or notified, never missed. + bundleContext.addBundleListener(this); + + // Load predefined items from this bundle, then from the bundles already started + loadPredefinedItems(bundleContext); for (Bundle bundle : bundleContext.getBundles()) { if (bundle.getBundleContext() != null && bundle.getBundleId() != bundleContext.getBundle().getBundleId()) { @@ -181,8 +191,6 @@ public abstract class AbstractMultiTypeCachingService extends AbstractContextAwa } } - bundleContext.addBundleListener(this); - // Load initial data for all types before starting timers loadInitialDataForAllTypes(); @@ -218,6 +226,7 @@ public abstract class AbstractMultiTypeCachingService extends AbstractContextAwa */ public void preDestroy() { bundleContext.removeBundleListener(this); + loadedBundleIds.clear(); shutdownTimers(); logger.debug("{} service shutdown.", getClass().getSimpleName()); } @@ -600,11 +609,13 @@ public abstract class AbstractMultiTypeCachingService extends AbstractContextAwa /** * Loads predefined JSON items from bundle resources for all configured types. + * Does nothing if the bundle's items are already loaded and the bundle has not stopped since. * * @param bundleContext the bundle context used to discover predefined resources */ protected void loadPredefinedItems(BundleContext bundleContext) { if (bundleContext == null) return; + if (!loadedBundleIds.add(bundleContext.getBundle().getBundleId())) return; for (CacheableTypeConfig<?> config : getTypeConfigs()) { if (config.hasPredefinedItems()) { @@ -812,6 +823,7 @@ public abstract class AbstractMultiTypeCachingService extends AbstractContextAwa // Clean up the tracking maps pluginContributions.remove(bundleId); pluginTypes.remove(bundleId); + loadedBundleIds.remove(bundleId); } } diff --git a/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java b/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java index 4729badf3..03dfbdb27 100644 --- a/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java +++ b/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java @@ -31,13 +31,21 @@ import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; +import org.osgi.framework.Bundle; +import org.osgi.framework.BundleContext; +import org.osgi.framework.BundleEvent; +import org.osgi.framework.BundleListener; +import java.io.File; import java.io.Serializable; +import java.net.URL; import java.util.*; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; +import java.util.function.Supplier; import static org.junit.Assert.*; import static org.mockito.ArgumentMatchers.*; @@ -585,6 +593,110 @@ public class AbstractMultiTypeCachingServiceTest { assertFalse(failure.getValue(), failure.getValue().contains(SYSTEM_TENANT)); } + /** + * UNOMI-1028: a plugin bundle that is not yet started when the startup scan reaches it, and starts + * before the scan ends, must still have its predefined items loaded. + */ + @Test + public void testBundleStartingDuringStartupScanIsLoaded() throws Exception { + BundleLoadingService service = newBundleLoadingService(); + BundleContext pluginContext = pluginContext(); + Bundle plugin = pluginContext.getBundle(); + AtomicBoolean started = new AtomicBoolean(); + when(plugin.getBundleContext()).thenAnswer(invocation -> { + if (started.compareAndSet(false, true)) { + // The scan sees the bundle still resolved; it starts right after. + for (BundleListener listener : service.listeners) { + listener.bundleChanged(new BundleEvent(BundleEvent.STARTED, plugin)); + } + return null; + } + return pluginContext; + }); + Bundle serviceBundle = service.serviceContext.getBundle(); + when(service.serviceContext.getBundles()).thenReturn(new Bundle[] {serviceBundle, plugin}); + + service.postConstruct(); + + verify(cacheService).put(eq(TEST_ITEM_TYPE), eq("predefined"), eq(SYSTEM_TENANT), any(TestSerializable.class)); + } + + /** A bundle seen by both the startup scan and the listener loads once, and again after a restart. */ + @Test + public void testBundleSeenByScanAndListenerLoadsOnceUntilRestarted() throws Exception { + BundleLoadingService service = newBundleLoadingService(); + BundleContext pluginContext = pluginContext(); + Bundle plugin = pluginContext.getBundle(); + when(plugin.getBundleContext()).thenReturn(pluginContext); + Bundle serviceBundle = service.serviceContext.getBundle(); + when(service.serviceContext.getBundles()).thenReturn(new Bundle[] {serviceBundle, plugin}); + + service.postConstruct(); + // The bundle was STARTING during the scan, so its STARTED event follows. + service.bundleChanged(new BundleEvent(BundleEvent.STARTED, plugin)); + verify(cacheService, times(1)).put(eq(TEST_ITEM_TYPE), eq("predefined"), eq(SYSTEM_TENANT), any(TestSerializable.class)); + + service.bundleChanged(new BundleEvent(BundleEvent.STOPPING, plugin)); + service.bundleChanged(new BundleEvent(BundleEvent.STARTED, plugin)); + verify(cacheService, times(2)).put(eq(TEST_ITEM_TYPE), eq("predefined"), eq(SYSTEM_TENANT), any(TestSerializable.class)); + } + + /** Service whose only type loads one predefined item per bundle, from {@code META-INF/cxs/test}. */ + private static class BundleLoadingService extends AbstractMultiTypeCachingService { + final BundleContext serviceContext = mock(BundleContext.class); + final List<BundleListener> listeners = new CopyOnWriteArrayList<>(); + + BundleLoadingService() { + Bundle serviceBundle = mock(Bundle.class); + when(serviceBundle.getBundleId()).thenReturn(1L); + when(serviceBundle.getBundleContext()).thenReturn(serviceContext); + when(serviceContext.getBundle()).thenReturn(serviceBundle); + doAnswer(invocation -> listeners.add(invocation.getArgument(0))) + .when(serviceContext).addBundleListener(any(BundleListener.class)); + setBundleContext(serviceContext); + } + + @Override + protected Set<CacheableTypeConfig<?>> getTypeConfigs() { + return Collections.singleton(CacheableTypeConfig.<TestSerializable>builder(TestSerializable.class, TEST_ITEM_TYPE, "test") + .withIdExtractor(TestSerializable::getId) + .withStreamProcessor((bundleContext, url, inputStream) -> new TestSerializable("predefined", null)) + .build()); + } + + @Override + protected void loadInitialDataForAllTypes() { + } + + @Override + protected void initializeTimers() { + } + } + + private BundleLoadingService newBundleLoadingService() { + doAnswer(invocation -> ((Supplier<?>) invocation.getArgument(0)).get()) + .when(contextManager).executeAsSystem(any(Supplier.class)); + BundleLoadingService service = new BundleLoadingService(); + service.setContextManager(contextManager); + service.setCacheService(cacheService); + service.setPersistenceService(persistenceService); + return service; + } + + /** Context of a plugin bundle (id 2) shipping one predefined item. */ + private BundleContext pluginContext() throws Exception { + File json = File.createTempFile("predefined", ".json"); + json.deleteOnExit(); + URL url = json.toURI().toURL(); + Bundle plugin = mock(Bundle.class); + when(plugin.getBundleId()).thenReturn(2L); + when(plugin.findEntries(eq("META-INF/cxs/test"), eq("*.json"), eq(true))) + .thenAnswer(invocation -> Collections.enumeration(Collections.singletonList(url))); + BundleContext pluginContext = mock(BundleContext.class); + when(pluginContext.getBundle()).thenReturn(plugin); + return pluginContext; + } + /** Records the tenants the action runs under; switching to {@code failingTenant} throws instead. */ private void recordTenantsRun(List<String> tenantsRun, String failingTenant) { doAnswer(invocation -> {
