This is an automated email from the ASF dual-hosted git repository.
jbonofre pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-karaf.git
The following commit(s) were added to refs/heads/main by this push:
new 9b4a09aaf fix(#734): build the type converter registry once, and say
when it is discarded (#739)
9b4a09aaf is described below
commit 9b4a09aafff8f30922ac2e170646601bdbe73640
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Sep 1 21:23:39 2026 +0200
fix(#734): build the type converter registry once, and say when it is
discarded (#739)
* fix(#734): build the type converter registry once, and say when it is
discarded
getDelegate() checked the volatile delegate for null and assigned it
without holding a lock. The volatile makes the read safe but not the
check-and-assign, so two threads racing on first access each built a
registry; one was returned, the other silently dropped along with
anything registered on it. Use double checked locking so the conversion
hot path stays lock free while first access happens once.
removedService and addingService now take the same lock, so a delegate
cannot be stopped and nulled while another thread is handing it out.
removedService also logs at WARN when it discards the delegate.
createRegistry replays the core converters and the loaders the tracker
still holds, but it cannot replay converters registered programmatically
via addTypeConverter or a Blueprint bean implementing TypeConverters, so
a single loader going away silently removes them from a running context.
Surfacing it does not fix the loss, but it stops it being invisible.
The asymmetry with addingService is deliberate and is left alone:
d54f9a806 (#625, PR #684) reverted invalidate-on-add to loading into the
existing delegate precisely to preserve programmatic registrations.
Actually replaying them on rebuild means recording them as they are
added, which is a larger design change than this.
The concurrency test fails 3 out of 3 runs against the previous
getDelegate and passes against this one, so it pins the behaviour rather
than being a flaky probe.
* fix(#734): address review - drop the tracker callback locking and replay
programmatic converters
Review feedback from jbonofre on #739.
Locking. addingService and removedService are no longer synchronized, and
createRegistry no longer calls back into the ServiceTracker at all: the
callbacks maintain a ConcurrentHashMap of the loaders the tracker hands us,
and the rebuild iterates a sorted snapshot of that map. Nothing calls into
the tracker or the framework while holding this instance's monitor any more,
so there is no lock ordering left to get wrong, and a slow loader.load() no
longer blocks every conversion in the container.
getDelegate is now plainly synchronized rather than double checked. Note
this
narrows but does not close the window in which a delegate handed to a caller
is stopped by removedService; that race predates this change and closing it
means not stopping the outgoing delegate eagerly.
Service use count. addingService takes the service, so releasing it is ours.
It now ungets on the failure path - the tracker treats a customizer that
throws as never tracked, so removedService would never run for that
reference
- and removedService ungets on the normal path, which leaked just as much.
Both leaks predate this PR.
Converter loss. Rather than only warning about it, registrations made
through
this facade (addTypeConverter, addTypeConverters, addBulkTypeConverters,
addFallbackTypeConverter, addConverter, removeTypeConverter, and the
setInjector / setTypeConverterExists / setTypeConverterExistsLoggingLevel
setters, which were lost the same way) are recorded in order and replayed
onto
the rebuilt registry. Removals replay too, so a rebuild reproduces the
sequence instead of resurrecting a converter that was deliberately taken
out.
Partial removal was the alternative, but attributing a converter to the
loader
that contributed it is not something TypeConverterRegistry exposes, and it
would undo the invalidate-and-rebuild strategy from d54f9a806 (#625).
The warning stays, reworded to say what actually happens, and is suppressed
while the context is stopping so tracker.close() no longer logs one per
loader
on shutdown.
Six new tests; four of them fail against the previous commit.
addingServiceMustNotHoldTheInstanceMonitor is a forward guard rather than a
regression test - the old lock free read in getDelegate meant it passed
there
too.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
* fix(#734): address second review - key the replay collection and make
registration atomic
Second review round from jbonofre on #739. He confirmed the AbstractTracked
correction and both ungetService fixes, and found four blocking items by
writing probe tests against the previous head. Three of them are fixed here;
the fourth is going to a follow-up issue at his suggestion.
register() and removeTypeConverter() are synchronized, on the monitor
getDelegate() already takes. Applying a registration and recording it had to
become one step: a rebuild landing between the two halves would replay a
collection the registration was not in yet and then hand back a registry it
had
never been applied to, so the converter would be recorded but absent - the
same
loss this PR exists to prevent, one level down. The monitor is reentrant, so
the nested getDelegate() costs nothing.
The replay collection is now keyed and insertion ordered rather than an
append-only list. The key is what the registration is about: a
TypeConvertible
for a converter, the contributed instance for the bulk, fallback and
addTypeConverters forms, a sentinel per setter. Re-registering replaces
instead
of appending, which is what a Blueprint container refresh does, and
removeTypeConverter deletes the matching entry instead of appending an
inverse.
The inverse reproduced the right end state on replay but kept the removed
converter - and the classloader of the bundle that contributed it - strongly
reachable for the life of the context, which is the leak shape that matters
in
OSGi. 500 add/remove pairs left 1000 entries before this; they leave none
now.
A removal for a pair that was never registered is no longer retained at all.
doStop() clears the collection. ServiceSupport permits stop then start, and
this is a context scoped service, so without it a restart replayed the
previous
lifecycle's registrations - including converters owned by bundles that are
gone
by the time the context comes back up.
Five new tests, sixteen in the class. Four of the five fail against the
previous commit, addRemovePairsDoNotAccumulate at exactly the 1000 entries
the
review reported. registrationCannotInterleaveWithARebuild passes there too,
so
it is a forward guard rather than a regression test: the old register()
still
reached a synchronized getDelegate(), so it blocked anyway. Reproducing the
actual interleave needs a registry that can be stalled inside its
addTypeConverter, as the review did with a throwaway.
Not fixed here, going to a follow-up issue: removedService writes delegate
outside the monitor, so an invalidation can be swallowed by a rebuild that
is
already in flight, and the wider monitor makes that window bigger rather
than
smaller. It is the same underlying problem as the eager stop of a delegate
callers already hold, and both belong in one issue.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../apache/camel/karaf/core/OsgiTypeConverter.java | 182 ++++++++++++---
.../camel/karaf/core/OsgiTypeConverterTest.java | 254 +++++++++++++++++++++
2 files changed, 407 insertions(+), 29 deletions(-)
diff --git
a/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java
b/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java
index 2aa75037c..b177e7074 100644
---
a/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java
+++
b/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java
@@ -20,8 +20,12 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
@@ -57,6 +61,32 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
private CamelContext camelContext;
private final Injector injector;
private final ServiceTracker<TypeConverterLoader, Object> tracker;
+ /**
+ * The loaders the tracker has handed us, kept here rather than read back
from the tracker: resolving them
+ * through the tracker inside {@link #createRegistry()} would mean calling
into the ServiceTracker and the
+ * framework while holding this instance's monitor.
+ */
+ private final Map<ServiceReference<TypeConverterLoader>,
TypeConverterLoader> trackedLoaders
+ = new ConcurrentHashMap<>();
+ /**
+ * Registrations made through this facade rather than by a {@link
TypeConverterLoader}, so a rebuilt registry can
+ * be brought back to the same state. Discarding the delegate would
otherwise drop them with no way to get them
+ * back.
+ * <p/>
+ * Keyed rather than a plain list, and insertion ordered so replay keeps
the original order. The key is what the
+ * registration is <em>about</em> - a {@link TypeConvertible} for a
converter, the contributed instance for the
+ * bulk and fallback forms, a sentinel for each setter - so re-registering
replaces instead of appending, and
+ * removing deletes the entry instead of appending an inverse. Both matter
in OSGi: each registration strongly
+ * references the converter it captured, and through it the classloader of
the bundle that contributed it, so a
+ * collection that only ever grows pins bundles that have long since been
uninstalled.
+ * <p/>
+ * Guarded by this instance's monitor, the same one {@link #getDelegate()}
takes.
+ */
+ private final Map<Object, Consumer<TypeConverterRegistry>>
programmaticRegistrations = new LinkedHashMap<>();
+
+ private static final Object INJECTOR_KEY = new Object();
+ private static final Object TYPE_CONVERTER_EXISTS_KEY = new Object();
+ private static final Object TYPE_CONVERTER_EXISTS_LOGGING_LEVEL_KEY = new
Object();
private volatile DefaultTypeConverter delegate;
private volatile boolean trackerOpened;
@@ -67,27 +97,36 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
this.tracker = new ServiceTracker<>(bundleContext,
TypeConverterLoader.class.getName(), this);
}
- private void ensureTrackerOpen() {
+ private synchronized void ensureTrackerOpen() {
if (!trackerOpened) {
tracker.open();
trackerOpened = true;
}
}
+ // deliberately not synchronized: the tracker calls this from the
framework's service event dispatch, and
+ // taking this instance's monitor here would put our lock on the far side
of the framework's, which is the
+ // ordering that makes a lock inversion possible
@Override
public Object addingService(ServiceReference<TypeConverterLoader>
serviceReference) {
LOG.trace("AddingService: {}, Bundle: {}", serviceReference,
serviceReference.getBundle());
TypeConverterLoader loader =
bundleContext.getService(serviceReference);
if (loader != null) {
+ trackedLoaders.put(serviceReference, loader);
try {
LOG.debug("loading type converter from bundle: {}",
serviceReference.getBundle().getSymbolicName());
- if (delegate != null) {
+ DefaultTypeConverter current = delegate;
+ if (current != null) {
// load the converter directly into the existing delegate
to preserve
// any converters that were added programmatically (e.g.
via Blueprint beans
// implementing TypeConverters)
- loader.load(delegate);
+ loader.load(current);
}
} catch (Throwable t) {
+ // the tracker treats a customizer that throws as "never
tracked", so it will not call
+ // removedService for this reference and nothing else will
release the use count taken above
+ trackedLoaders.remove(serviceReference);
+ ungetQuietly(serviceReference);
throw new RuntimeCamelException("Error loading type converters
from service: " + serviceReference + " due: " + t.getMessage(), t);
}
}
@@ -99,9 +138,21 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
public void modifiedService(ServiceReference<TypeConverterLoader>
serviceReference, Object o) {
}
+ // not synchronized, for the same reason as addingService
@Override
public void removedService(ServiceReference<TypeConverterLoader>
serviceReference, Object o) {
LOG.trace("RemovedService: {}, Bundle: {}", serviceReference,
serviceReference.getBundle());
+ trackedLoaders.remove(serviceReference);
+ // we took the service in addingService, so releasing it is ours to do
+ ungetQuietly(serviceReference);
+ if (this.delegate != null && !isStopping() && !isStopped()) {
+ // worth saying out loud: one loader going away discards the whole
registry, and the rebuild is a
+ // full reload of the core converters plus every remaining loader,
not an incremental removal
+ LOG.warn("TypeConverterLoader from bundle {} was unregistered,
discarding the type converter registry;"
+ + " it is rebuilt on next use from the remaining loaders,
and the converters registered"
+ + " programmatically on this context are replayed onto
the rebuilt registry.",
+ serviceReference.getBundle() != null ?
serviceReference.getBundle().getSymbolicName() : serviceReference);
+ }
try {
ServiceHelper.stopService(this.delegate);
} catch (Exception e) {
@@ -112,6 +163,16 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
this.delegate = null;
}
+ private void ungetQuietly(ServiceReference<TypeConverterLoader>
serviceReference) {
+ try {
+ bundleContext.ungetService(serviceReference);
+ } catch (Exception e) {
+ // the bundle or the framework may already be gone; releasing the
use count is best effort
+ LOG.debug("Error ungetting service {} due: {}. This exception will
be ignored.", serviceReference,
+ e.getMessage(), e);
+ }
+ }
+
@Override
protected void doStart() throws Exception {
ensureTrackerOpen();
@@ -121,6 +182,14 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
protected void doStop() throws Exception {
this.tracker.close();
this.trackerOpened = false;
+ // close() calls removedService for everything still tracked, this
only makes the end state explicit
+ this.trackedLoaders.clear();
+ synchronized (this) {
+ // ServiceSupport allows stop then start again, and this is a
context scoped service, so without this a
+ // restart would replay the previous lifecycle's registrations -
including converters belonging to
+ // bundles that are gone by the time the context comes back up
+ this.programmaticRegistrations.clear();
+ }
ServiceHelper.stopService(this.delegate);
this.delegate = null;
}
@@ -173,27 +242,33 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
@Override
public void addTypeConverter(Class<?> toType, Class<?> fromType,
TypeConverter typeConverter) {
- getDelegate().addTypeConverter(toType, fromType, typeConverter);
+ register(new TypeConvertible<>(fromType, toType),
+ registry -> registry.addTypeConverter(toType, fromType,
typeConverter));
}
@Override
public void addTypeConverters(Object typeConverters) {
- getDelegate().addTypeConverters(typeConverters);
+ register(typeConverters, registry ->
registry.addTypeConverters(typeConverters));
}
@Override
public void addBulkTypeConverters(BulkTypeConverters bulkTypeConverters) {
- getDelegate().addBulkTypeConverters(bulkTypeConverters);
+ register(bulkTypeConverters, registry ->
registry.addBulkTypeConverters(bulkTypeConverters));
}
@Override
- public boolean removeTypeConverter(Class<?> toType, Class<?> fromType) {
- return getDelegate().removeTypeConverter(toType, fromType);
+ public synchronized boolean removeTypeConverter(Class<?> toType, Class<?>
fromType) {
+ boolean removed = getDelegate().removeTypeConverter(toType, fromType);
+ // delete the matching registration rather than recording an inverse.
An inverse would reproduce the same
+ // end state on replay, but it would also keep the removed converter -
and its bundle's classloader -
+ // strongly reachable for the life of the context, and cost a no-op
call on every future rebuild
+ programmaticRegistrations.remove(new TypeConvertible<>(fromType,
toType));
+ return removed;
}
@Override
public void addFallbackTypeConverter(TypeConverter typeConverter, boolean
canPromote) {
- getDelegate().addFallbackTypeConverter(typeConverter, canPromote);
+ register(typeConverter, registry ->
registry.addFallbackTypeConverter(typeConverter, canPromote));
}
@Override
@@ -203,7 +278,7 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
@Override
public void setInjector(Injector injector) {
- getDelegate().setInjector(injector);
+ register(INJECTOR_KEY, registry -> registry.setInjector(injector));
}
@Override
@@ -228,7 +303,8 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
@Override
public void setTypeConverterExistsLoggingLevel(LoggingLevel loggingLevel) {
- getDelegate().setTypeConverterExistsLoggingLevel(loggingLevel);
+ register(TYPE_CONVERTER_EXISTS_LOGGING_LEVEL_KEY,
+ registry ->
registry.setTypeConverterExistsLoggingLevel(loggingLevel));
}
@Override
@@ -238,10 +314,13 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
@Override
public void setTypeConverterExists(TypeConverterExists
typeConverterExists) {
- getDelegate().setTypeConverterExists(typeConverterExists);
+ register(TYPE_CONVERTER_EXISTS_KEY, registry ->
registry.setTypeConverterExists(typeConverterExists));
}
- public DefaultTypeConverter getDelegate() {
+ // fully synchronized rather than double checked: the delegate is not
immutable after publication -
+ // removedService stops and replaces it - so a lock free read of the field
buys a race for no real gain,
+ // conversion work dwarfing an uncontended monitor either way
+ public synchronized DefaultTypeConverter getDelegate() {
if (delegate == null) {
// ensure the tracker is open so we can discover
TypeConverterLoader services
// before creating the registry - this is important because
getDelegate() may be
@@ -279,28 +358,73 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
throw new RuntimeCamelException("Error loading CoreTypeConverter
due: " + e.getMessage(), e);
}
- // Load the type converters the tracker has been tracking
- // Here we need to use the ServiceReference to check the ranking
- ServiceReference<TypeConverterLoader>[] serviceReferences =
this.tracker.getServiceReferences();
- if (serviceReferences != null) {
- ArrayList<ServiceReference<TypeConverterLoader>> servicesList =
- new ArrayList<>(Arrays.asList(serviceReferences));
- // Just make sure we install the high ranking fallback converter
at last
- Collections.sort(servicesList);
- for (ServiceReference<TypeConverterLoader> sr : servicesList) {
- try {
- LOG.debug("loading type converter from bundle: {}",
sr.getBundle().getSymbolicName());
-
((TypeConverterLoader)this.tracker.getService(sr)).load(answer);
- } catch (Throwable t) {
- throw new RuntimeCamelException("Error loading type
converters from service: " + sr + " due: " + t.getMessage(), t);
- }
+ // Load the type converters the tracker has been tracking. These come
from our own map rather than from
+ // tracker.getServiceReferences()/getService(): this runs while
holding this instance's monitor, and
+ // calling back into the tracker from here is what would establish a
lock ordering against the framework.
+ List<ServiceReference<TypeConverterLoader>> servicesList = new
ArrayList<>(trackedLoaders.keySet());
+ // Just make sure we install the high ranking fallback converter at
last
+ Collections.sort(servicesList);
+ for (ServiceReference<TypeConverterLoader> sr : servicesList) {
+ TypeConverterLoader loader = trackedLoaders.get(sr);
+ if (loader == null) {
+ // unregistered between the snapshot and here
+ continue;
+ }
+ try {
+ LOG.debug("loading type converter from bundle: {}",
sr.getBundle().getSymbolicName());
+ loader.load(answer);
+ } catch (Throwable t) {
+ throw new RuntimeCamelException("Error loading type converters
from service: " + sr + " due: " + t.getMessage(), t);
}
}
+ replayProgrammaticRegistrations(answer);
+
LOG.trace("Created TypeConverter: {}", answer);
return answer;
}
+ /**
+ * Re-applies everything that was registered through this facade rather
than by a
+ * {@link TypeConverterLoader}, in the order it was originally applied.
+ */
+ private void replayProgrammaticRegistrations(DefaultTypeConverter
registry) {
+ if (programmaticRegistrations.isEmpty()) {
+ return;
+ }
+ LOG.debug("Replaying {} programmatic registration(s) onto the rebuilt
type converter registry",
+ programmaticRegistrations.size());
+ for (Consumer<TypeConverterRegistry> registration :
programmaticRegistrations.values()) {
+ registration.accept(registry);
+ }
+ }
+
+ /**
+ * Applies a registration to the current delegate and remembers it under
{@code key}, so that discarding the
+ * delegate does not discard the registration with it.
+ * <p/>
+ * Synchronized, and on the same monitor {@link #getDelegate()} takes:
applying and recording have to be one
+ * step. A rebuild landing between them would replay a list this
registration is not in yet and then hand back
+ * a registry it was never applied to, so the converter would be recorded
but not actually present - the same
+ * loss this class is meant to prevent, one level down. The monitor is
reentrant, so the nested
+ * {@link #getDelegate()} is free.
+ */
+ /**
+ * How many programmatic registrations are currently retained for replay.
Package private for the tests, which
+ * pin the invariant that this does not grow without bound - each entry
retains the converter it captured, and
+ * with it the classloader of the contributing bundle.
+ */
+ synchronized int programmaticRegistrationCount() {
+ return programmaticRegistrations.size();
+ }
+
+ private synchronized void register(Object key,
Consumer<TypeConverterRegistry> registration) {
+ // apply first: a registration the delegate rejects is not one worth
replaying. getDelegate() may build the
+ // registry here, replaying the map as it stands - this entry goes in
after, so it cannot be applied twice
+ registration.accept(getDelegate());
+ programmaticRegistrations.put(key, registration);
+ }
+
private class OsgiDefaultTypeConverter extends DefaultTypeConverter {
public OsgiDefaultTypeConverter(PackageScanClassResolver resolver,
Injector injector, boolean loadTypeConverters,
@@ -332,7 +456,7 @@ public class OsgiTypeConverter extends ServiceSupport
implements TypeConverter,
@Override
public void addConverter(TypeConvertible<?, ?> typeConvertible,
TypeConverter typeConverter) {
- getDelegate().addConverter(typeConvertible, typeConverter);
+ register(typeConvertible, registry ->
registry.addConverter(typeConvertible, typeConverter));
}
}
\ No newline at end of file
diff --git
a/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java
b/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java
index 838193400..92f9b89c4 100644
---
a/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java
+++
b/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java
@@ -17,7 +17,20 @@
package org.apache.camel.karaf.core;
import org.apache.camel.CamelContext;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.TypeConverter;
import org.apache.camel.spi.Injector;
+import java.util.ArrayList;
+import java.util.List;
+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.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.impl.converter.DefaultTypeConverter;
import org.apache.camel.spi.TypeConverterLoader;
import org.apache.camel.spi.TypeConverterRegistry;
import org.junit.jupiter.api.BeforeEach;
@@ -46,6 +59,8 @@ public class OsgiTypeConverterTest {
@Mock
private TypeConverterLoader loader;
@Mock
+ private TypeConverter typeConverter;
+ @Mock
private Bundle bundle;
private OsgiTypeConverter osgiTypeConverter;
@@ -110,4 +125,243 @@ public class OsgiTypeConverterTest {
var delegateAfter = osgiTypeConverter.getDelegate();
assertNotNull(delegateAfter);
}
+
+ @Test
+ void concurrentFirstAccessShouldBuildTheRegistryOnce() throws Exception {
+ int threads = 16;
+ AtomicInteger created = new AtomicInteger();
+ CountDownLatch startLine = new CountDownLatch(1);
+
+ OsgiTypeConverter counting = new OsgiTypeConverter(bundleContext,
camelContext, injector) {
+ @Override
+ protected DefaultTypeConverter createRegistry() {
+ created.incrementAndGet();
+ return super.createRegistry();
+ }
+ };
+
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ List<Future<DefaultTypeConverter>> futures = new ArrayList<>();
+ for (int i = 0; i < threads; i++) {
+ futures.add(pool.submit(() -> {
+ startLine.await();
+ return counting.getDelegate();
+ }));
+ }
+ // release them all at once so they race on the null check
+ startLine.countDown();
+
+ DefaultTypeConverter first = futures.get(0).get(30,
TimeUnit.SECONDS);
+ assertNotNull(first);
+ for (Future<DefaultTypeConverter> f : futures) {
+ assertSame(first, f.get(30, TimeUnit.SECONDS),
+ "every caller must see the same registry instance");
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+
+ assertEquals(1, created.get(),
+ "the registry must be built exactly once, otherwise converters
registered on a discarded"
+ + " instance are silently lost");
+ }
+
+ /** Marker source type, so the registered converter cannot collide with a
core one. */
+ interface Marker {
+ }
+
+ @Test
+ void rebuiltRegistryReloadsTheTrackedLoadersWithoutAskingTheTracker()
throws Exception {
+ // arrives before the registry exists, so addingService only records it
+ osgiTypeConverter.addingService(serviceReference);
+ verify(loader, never()).load(any());
+
+ DefaultTypeConverter first = osgiTypeConverter.getDelegate();
+
+ // createRegistry replayed it from the recorded loaders; it never
called back into the ServiceTracker,
+ // which is what would put a framework call underneath this instance's
monitor
+ verify(loader).load(first);
+ }
+
+ @Test
+ void programmaticConverterSurvivesARegistryRebuild() throws Exception {
+ DefaultTypeConverter before = osgiTypeConverter.getDelegate();
+ osgiTypeConverter.addTypeConverter(String.class, Marker.class,
typeConverter);
+ assertNotNull(before.lookup(String.class, Marker.class),
"precondition: the converter is registered");
+
+ // one loader going away discards the whole registry
+ osgiTypeConverter.removedService(serviceReference, loader);
+ DefaultTypeConverter after = osgiTypeConverter.getDelegate();
+
+ assertNotSame(before, after, "the registry should have been rebuilt");
+ assertNotNull(after.lookup(String.class, Marker.class),
+ "a converter registered programmatically must be replayed onto
the rebuilt registry, otherwise it"
+ + " disappears from a running context when any bundle
unregisters a loader");
+ }
+
+ @Test
+ void removedTypeConverterIsNotResurrectedByARebuild() throws Exception {
+ osgiTypeConverter.addTypeConverter(String.class, Marker.class,
typeConverter);
+ osgiTypeConverter.removeTypeConverter(String.class, Marker.class);
+
+ osgiTypeConverter.removedService(serviceReference, loader);
+
+ assertNull(osgiTypeConverter.getDelegate().lookup(String.class,
Marker.class),
+ "the replay must reproduce the sequence, not just the
additions");
+ }
+
+ @Test
+ void addingServiceReleasesTheServiceWhenLoadingFails() throws Exception {
+ osgiTypeConverter.getDelegate();
+ doThrow(new RuntimeException("boom")).when(loader).load(any());
+
+ assertThrows(RuntimeCamelException.class, () ->
osgiTypeConverter.addingService(serviceReference));
+
+ // a customizer that throws is treated as never tracked, so
removedService will not run for this
+ // reference and nothing else would release the use count taken by
addingService
+ verify(bundleContext).ungetService(serviceReference);
+ }
+
+ @Test
+ void removedServiceReleasesTheService() {
+ osgiTypeConverter.addingService(serviceReference);
+
+ osgiTypeConverter.removedService(serviceReference, loader);
+
+ verify(bundleContext).ungetService(serviceReference);
+ }
+
+ @Test
+ void addingServiceMustNotHoldTheInstanceMonitor() throws Exception {
+ // build first, so addingService takes the branch that calls into the
loader
+ osgiTypeConverter.getDelegate();
+
+ CountDownLatch insideLoad = new CountDownLatch(1);
+ CountDownLatch releaseLoad = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ insideLoad.countDown();
+ releaseLoad.await(30, TimeUnit.SECONDS);
+ return null;
+ }).when(loader).load(any());
+
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ try {
+ Future<?> adding = pool.submit(() ->
osgiTypeConverter.addingService(serviceReference));
+ assertTrue(insideLoad.await(30, TimeUnit.SECONDS), "addingService
should have reached loader.load");
+
+ // the framework calls addingService while it is dispatching a
service event; if it took this
+ // instance's monitor, every conversion in the container would
block behind an arbitrary bundle's
+ // loader for as long as that loader takes
+ Future<DefaultTypeConverter> reader =
pool.submit(osgiTypeConverter::getDelegate);
+ assertNotNull(reader.get(10, TimeUnit.SECONDS),
+ "getDelegate must not be blocked by an in-flight
addingService");
+
+ releaseLoad.countDown();
+ adding.get(30, TimeUnit.SECONDS);
+ } finally {
+ releaseLoad.countDown();
+ pool.shutdownNow();
+ }
+ }
+
+ @Test
+ void addRemovePairsDoNotAccumulate() {
+ osgiTypeConverter.getDelegate();
+
+ for (int i = 0; i < 500; i++) {
+ osgiTypeConverter.addTypeConverter(String.class, Marker.class,
typeConverter);
+ osgiTypeConverter.removeTypeConverter(String.class, Marker.class);
+ }
+
+ // an inverse-appending list would sit at 1000 here, and every
converter it captured - and the classloader
+ // of the bundle that contributed it - would stay strongly reachable
for the life of the context
+ assertEquals(0, osgiTypeConverter.programmaticRegistrationCount(),
+ "add/remove pairs must prune, not accumulate");
+ }
+
+ @Test
+ void reRegisteringTheSameConversionReplaces() {
+ osgiTypeConverter.getDelegate();
+
+ for (int i = 0; i < 10; i++) {
+ // what a Blueprint container refresh looks like: the same
conversion registered again
+ osgiTypeConverter.addTypeConverter(String.class, Marker.class,
typeConverter);
+ }
+
+ assertEquals(1, osgiTypeConverter.programmaticRegistrationCount(),
+ "re-registering the same conversion must replace rather than
append");
+ }
+
+ @Test
+ void removingAConversionThatWasNeverRegisteredDoesNotAccumulate() {
+ osgiTypeConverter.getDelegate();
+
+ osgiTypeConverter.removeTypeConverter(String.class, Marker.class);
+
+ assertEquals(0, osgiTypeConverter.programmaticRegistrationCount(),
+ "a removal for a pair that was never registered must not be
retained");
+ }
+
+ @Test
+ void restartDoesNotReplayThePreviousLifecycle() throws Exception {
+ osgiTypeConverter.start();
+ osgiTypeConverter.addTypeConverter(String.class, Marker.class,
typeConverter);
+ assertNotNull(osgiTypeConverter.getDelegate().lookup(String.class,
Marker.class));
+
+ osgiTypeConverter.stop();
+ osgiTypeConverter.start();
+
+ // the converters captured before the stop belong to bundles that may
be gone by now
+ assertEquals(0, osgiTypeConverter.programmaticRegistrationCount());
+ assertNull(osgiTypeConverter.getDelegate().lookup(String.class,
Marker.class),
+ "a stop/start cycle must not resurrect the previous
lifecycle's registrations");
+ }
+
+ @Test
+ void registrationCannotInterleaveWithARebuild() throws Exception {
+ CountDownLatch insideBuild = new CountDownLatch(1);
+ CountDownLatch releaseBuild = new CountDownLatch(1);
+
+ OsgiTypeConverter stalling = new OsgiTypeConverter(bundleContext,
camelContext, injector) {
+ @Override
+ protected DefaultTypeConverter createRegistry() {
+ DefaultTypeConverter built = super.createRegistry();
+ insideBuild.countDown();
+ try {
+ releaseBuild.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return built;
+ }
+ };
+
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ try {
+ Future<DefaultTypeConverter> builder =
pool.submit(stalling::getDelegate);
+ assertTrue(insideBuild.await(30, TimeUnit.SECONDS), "the rebuild
should have started");
+
+ Future<?> registrar = pool.submit(() -> {
+ stalling.addTypeConverter(String.class, Marker.class,
typeConverter);
+ return null;
+ });
+
+ // apply-and-record has to be one step against the rebuild. If it
were not, this registration would be
+ // applied to the registry being discarded and only recorded
afterwards, so the rebuilt one would
+ // neither have it applied nor replay it
+ assertThrows(TimeoutException.class, () -> registrar.get(2,
TimeUnit.SECONDS),
+ "a registration must not proceed while a rebuild holds the
monitor");
+
+ releaseBuild.countDown();
+ builder.get(30, TimeUnit.SECONDS);
+ registrar.get(30, TimeUnit.SECONDS);
+
+ assertNotNull(stalling.getDelegate().lookup(String.class,
Marker.class),
+ "the registration must land on the registry that is
actually live");
+ } finally {
+ releaseBuild.countDown();
+ pool.shutdownNow();
+ }
+ }
}
