This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 1285eb670 UNOMI-1028: Load built-in types from bundles that start
during the startup scan (#933)
1285eb670 is described below
commit 1285eb670c511786f553dd1aa97f9811632537ae
Author: Serge Huber <[email protected]>
AuthorDate: Thu Oct 29 16:50:33 2026 +0100
UNOMI-1028: Load built-in types from bundles that start during the startup
scan (#933)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../cache/AbstractMultiTypeCachingService.java | 28 ++-
.../cache/AbstractMultiTypeCachingServiceTest.java | 224 +++++++++++++++++++++
2 files changed, 245 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..705d41dc2 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,13 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
*/
protected final Map<Long, List<PluginType>> pluginTypes = new
ConcurrentHashMap<>();
+ /**
+ * IDs of the bundles whose predefined items have started loading
(possibly still in progress on
+ * another thread). A bundle that starts during the startup scan may be
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 +170,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 +192,6 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
}
}
- bundleContext.addBundleListener(this);
-
// Load initial data for all types before starting timers
loadInitialDataForAllTypes();
@@ -218,6 +227,7 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
*/
public void preDestroy() {
bundleContext.removeBundleListener(this);
+ loadedBundleIds.clear();
shutdownTimers();
logger.debug("{} service shutdown.", getClass().getSimpleName());
}
@@ -600,11 +610,14 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
/**
* Loads predefined JSON items from bundle resources for all configured
types.
+ * Does nothing if an earlier call already started loading this bundle (it
may still be running on
+ * another thread) 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 +825,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..727f7ce64 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,26 @@ 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.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
+import java.util.function.Supplier;
import static org.junit.Assert.*;
import static org.mockito.ArgumentMatchers.*;
@@ -585,6 +598,217 @@ 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));
+ }
+
+ /**
+ * UNOMI-1028 with real threads: the startup scan runs on one thread while
a plugin bundle starts
+ * on another at a random moment, as the Blueprint extender and Karaf
feature threads do. Like the
+ * OSGi framework, the starter makes the bundle active before notifying
the listeners registered at
+ * that time. Every round must load the plugin's items exactly once.
+ */
+ @Test
+ public void
testBundleStartingConcurrentlyWithStartupScanLoadsExactlyOnce() throws
Exception {
+ final int rounds = 300;
+ // Bundles the scan walks after the plugin, about a microsecond each
(the real scan reads JSON),
+ // so that the plugin can start while the rest of the scan runs.
+ final int laterBundles = 200;
+
+ ExecutionContextManager systemContext =
mock(ExecutionContextManager.class, withSettings().stubOnly());
+ doAnswer(invocation -> ((Supplier<?>) invocation.getArgument(0)).get())
+ .when(systemContext).executeAsSystem(any(Supplier.class));
+ MultiTypeCacheService cache = mock(MultiTypeCacheService.class,
withSettings().stubOnly());
+
+ File jsonFile = File.createTempFile("predefined", ".json");
+ jsonFile.deleteOnExit();
+ URL json = jsonFile.toURI().toURL();
+ AtomicBoolean pluginStarted = new AtomicBoolean();
+ Bundle plugin = mock(Bundle.class, withSettings().stubOnly());
+ BundleContext pluginContext = mock(BundleContext.class,
withSettings().stubOnly());
+ when(pluginContext.getBundle()).thenReturn(plugin);
+ when(plugin.getBundleId()).thenReturn(2L);
+ when(plugin.getBundleContext()).thenAnswer(invocation ->
pluginStarted.get() ? pluginContext : null);
+ when(plugin.findEntries(eq("META-INF/cxs/test"), eq("*.json"),
eq(true)))
+ .thenAnswer(invocation ->
Collections.enumeration(Collections.singletonList(json)));
+ Bundle resolvedBundle = mock(Bundle.class, withSettings().stubOnly());
+ when(resolvedBundle.getBundleContext()).thenAnswer(invocation -> {
+ spinFor(1_000);
+ return null;
+ });
+
+ // One service, restarted every round: creating mocks per round would
dominate the run time.
+ BundleLoadingService service = new BundleLoadingService();
+ service.setContextManager(systemContext);
+ service.setCacheService(cache);
+ Bundle[] bundles = new Bundle[laterBundles + 2];
+ bundles[0] = service.serviceContext.getBundle();
+ bundles[1] = plugin;
+ Arrays.fill(bundles, 2, bundles.length, resolvedBundle);
+ AtomicBoolean scanStarted = new AtomicBoolean();
+ when(service.serviceContext.getBundles()).thenAnswer(invocation -> {
+ scanStarted.set(true);
+ return bundles;
+ });
+
+ ExecutorService starter = Executors.newSingleThreadExecutor();
+ int missed = 0;
+ int loadedTwice = 0;
+ try {
+ for (int round = 0; round < rounds; round++) {
+ service.preDestroy();
+ service.loads.set(0);
+ pluginStarted.set(false);
+ scanStarted.set(false);
+
+ // Start the plugin somewhere between the scan's start and
shortly after its end.
+ long delayNanos =
ThreadLocalRandom.current().nextLong(TimeUnit.MICROSECONDS.toNanos(300));
+ Future<?> start = starter.submit(() -> {
+ // Spin rather than block: waking a parked thread takes
longer than the scan.
+ while (!scanStarted.get()) {
+ Thread.onSpinWait();
+ }
+ spinFor(delayNanos);
+ pluginStarted.set(true);
+ for (BundleListener listener : service.listeners) {
+ listener.bundleChanged(new
BundleEvent(BundleEvent.STARTED, plugin));
+ }
+ return null;
+ });
+ service.postConstruct();
+ start.get(5, TimeUnit.SECONDS);
+
+ if (service.loads.get() == 0) {
+ missed++;
+ } else if (service.loads.get() > 1) {
+ loadedTwice++;
+ }
+ }
+ } finally {
+ starter.shutdownNow();
+ }
+
+ assertEquals("rounds where the plugin's items were never loaded", 0,
missed);
+ assertEquals("rounds where the plugin's items were loaded twice", 0,
loadedTwice);
+ }
+
+ private static void spinFor(long nanos) {
+ long until = System.nanoTime() + nanos;
+ while (System.nanoTime() < until) {
+ Thread.onSpinWait();
+ }
+ }
+
+ /** 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<>();
+ final AtomicInteger loads = new AtomicInteger();
+
+ 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));
+ doAnswer(invocation -> listeners.remove(invocation.getArgument(0)))
+
.when(serviceContext).removeBundleListener(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) -> {
+ loads.incrementAndGet();
+ return new TestSerializable("predefined", null);
+ })
+ .build());
+ }
+
+ @Override
+ protected void loadInitialDataForAllTypes() {
+ }
+
+ @Override
+ protected void initializeTimers() {
+ }
+
+ @Override
+ protected void shutdownTimers() {
+ }
+ }
+
+ 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 -> {
