From a8b61fbf81c54e01d3876901ca39bf4471300089 Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Mon, 24 Aug 2026 22:35:39 +0200 Subject: [PATCH 1/3] 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. --- .../camel/karaf/core/OsgiTypeConverter.java | 42 +++++++++++---- .../karaf/core/OsgiTypeConverterTest.java | 51 +++++++++++++++++++ 2 files changed, 82 insertions(+), 11 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..7f3334200 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 @@ -67,7 +67,7 @@ public OsgiTypeConverter(BundleContext bundleContext, CamelContext camelContext, this.tracker = new ServiceTracker<>(bundleContext, TypeConverterLoader.class.getName(), this); } - private void ensureTrackerOpen() { + private synchronized void ensureTrackerOpen() { if (!trackerOpened) { tracker.open(); trackerOpened = true; @@ -75,7 +75,7 @@ private void ensureTrackerOpen() { } @Override - public Object addingService(ServiceReference serviceReference) { + public synchronized Object addingService(ServiceReference serviceReference) { LOG.trace("AddingService: {}, Bundle: {}", serviceReference, serviceReference.getBundle()); TypeConverterLoader loader = bundleContext.getService(serviceReference); if (loader != null) { @@ -100,8 +100,17 @@ public void modifiedService(ServiceReference serviceReferen } @Override - public void removedService(ServiceReference serviceReference, Object o) { + public synchronized void removedService(ServiceReference serviceReference, Object o) { LOG.trace("RemovedService: {}, Bundle: {}", serviceReference, serviceReference.getBundle()); + if (this.delegate != null) { + // the rebuild in createRegistry replays the core converters and the loaders the tracker still + // holds, but it cannot replay converters that were registered programmatically (addTypeConverter, + // or a Blueprint bean implementing TypeConverters) - those are lost with the delegate + LOG.warn("TypeConverterLoader from bundle {} was unregistered, discarding and rebuilding the type" + + " converter registry. Type converters that were registered programmatically on the running" + + " context are not restored by the rebuild and have to be registered again.", + serviceReference.getBundle() != null ? serviceReference.getBundle().getSymbolicName() : serviceReference); + } try { ServiceHelper.stopService(this.delegate); } catch (Exception e) { @@ -242,15 +251,26 @@ public void setTypeConverterExists(TypeConverterExists typeConverterExists) { } public 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 - // called during doInit() (e.g. when to() eagerly creates endpoints) which happens - // before doStart() where the tracker is normally opened - ensureTrackerOpen(); - delegate = createRegistry(); + DefaultTypeConverter answer = delegate; + if (answer == null) { + // double checked locking against the volatile field: getDelegate is on the conversion hot path, + // so the common case must stay lock free, but the check and the assignment together are not + // atomic - without the lock two threads racing on first access each build a registry, and + // whatever was registered on the one that loses is silently dropped + synchronized (this) { + answer = delegate; + if (answer == null) { + // ensure the tracker is open so we can discover TypeConverterLoader services + // before creating the registry - this is important because getDelegate() may be + // called during doInit() (e.g. when to() eagerly creates endpoints) which happens + // before doStart() where the tracker is normally opened + ensureTrackerOpen(); + answer = createRegistry(); + delegate = answer; + } + } } - return delegate; + return answer; } protected DefaultTypeConverter createRegistry() { 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..3194c1a8e 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 @@ -18,6 +18,16 @@ import org.apache.camel.CamelContext; 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.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; @@ -110,4 +120,45 @@ void removedServiceShouldInvalidateDelegate() throws Exception { 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> 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 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"); + } } From 2c0676e260cf71024eb85f7ad43ae296e30f8b5f Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Tue, 25 Aug 2026 15:29:52 +0200 Subject: [PATCH 2/3] 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) --- .../camel/karaf/core/OsgiTypeConverter.java | 181 ++++++++++++------ .../karaf/core/OsgiTypeConverterTest.java | 102 ++++++++++ 2 files changed, 226 insertions(+), 57 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 7f3334200..5546799b3 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.Arrays; import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; import org.apache.camel.CamelContext; import org.apache.camel.Exchange; @@ -57,6 +61,19 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter, private CamelContext camelContext; private final Injector injector; private final ServiceTracker 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, TypeConverterLoader> trackedLoaders + = new ConcurrentHashMap<>(); + /** + * Registrations made through this facade rather than by a {@link TypeConverterLoader}, in the order they were + * made, 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. + */ + private final List> programmaticRegistrations = new CopyOnWriteArrayList<>(); private volatile DefaultTypeConverter delegate; private volatile boolean trackerOpened; @@ -74,20 +91,29 @@ private synchronized void ensureTrackerOpen() { } } + // 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 synchronized Object addingService(ServiceReference serviceReference) { + public Object addingService(ServiceReference 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,16 +125,19 @@ public synchronized Object addingService(ServiceReference s public void modifiedService(ServiceReference serviceReference, Object o) { } + // not synchronized, for the same reason as addingService @Override - public synchronized void removedService(ServiceReference serviceReference, Object o) { + public void removedService(ServiceReference serviceReference, Object o) { LOG.trace("RemovedService: {}, Bundle: {}", serviceReference, serviceReference.getBundle()); - if (this.delegate != null) { - // the rebuild in createRegistry replays the core converters and the loaders the tracker still - // holds, but it cannot replay converters that were registered programmatically (addTypeConverter, - // or a Blueprint bean implementing TypeConverters) - those are lost with the delegate - LOG.warn("TypeConverterLoader from bundle {} was unregistered, discarding and rebuilding the type" - + " converter registry. Type converters that were registered programmatically on the running" - + " context are not restored by the rebuild and have to be registered again.", + 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 { @@ -121,6 +150,16 @@ public synchronized void removedService(ServiceReference se this.delegate = null; } + private void ungetQuietly(ServiceReference 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(); @@ -130,6 +169,8 @@ protected void doStart() throws Exception { 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(); ServiceHelper.stopService(this.delegate); this.delegate = null; } @@ -182,27 +223,30 @@ public T tryConvertTo(Class type, Object value) { @Override public void addTypeConverter(Class toType, Class fromType, TypeConverter typeConverter) { - getDelegate().addTypeConverter(toType, fromType, typeConverter); + register(registry -> registry.addTypeConverter(toType, fromType, typeConverter)); } @Override public void addTypeConverters(Object typeConverters) { - getDelegate().addTypeConverters(typeConverters); + register(registry -> registry.addTypeConverters(typeConverters)); } @Override public void addBulkTypeConverters(BulkTypeConverters bulkTypeConverters) { - getDelegate().addBulkTypeConverters(bulkTypeConverters); + register(registry -> registry.addBulkTypeConverters(bulkTypeConverters)); } @Override public boolean removeTypeConverter(Class toType, Class fromType) { - return getDelegate().removeTypeConverter(toType, fromType); + boolean removed = getDelegate().removeTypeConverter(toType, fromType); + // replayed as well, so a rebuild reproduces the sequence rather than resurrecting the converter + programmaticRegistrations.add(registry -> registry.removeTypeConverter(toType, fromType)); + return removed; } @Override public void addFallbackTypeConverter(TypeConverter typeConverter, boolean canPromote) { - getDelegate().addFallbackTypeConverter(typeConverter, canPromote); + register(registry -> registry.addFallbackTypeConverter(typeConverter, canPromote)); } @Override @@ -212,7 +256,7 @@ public TypeConverter lookup(Class toType, Class fromType) { @Override public void setInjector(Injector injector) { - getDelegate().setInjector(injector); + register(registry -> registry.setInjector(injector)); } @Override @@ -237,7 +281,7 @@ public LoggingLevel getTypeConverterExistsLoggingLevel() { @Override public void setTypeConverterExistsLoggingLevel(LoggingLevel loggingLevel) { - getDelegate().setTypeConverterExistsLoggingLevel(loggingLevel); + register(registry -> registry.setTypeConverterExistsLoggingLevel(loggingLevel)); } @Override @@ -247,30 +291,22 @@ public TypeConverterExists getTypeConverterExists() { @Override public void setTypeConverterExists(TypeConverterExists typeConverterExists) { - getDelegate().setTypeConverterExists(typeConverterExists); - } - - public DefaultTypeConverter getDelegate() { - DefaultTypeConverter answer = delegate; - if (answer == null) { - // double checked locking against the volatile field: getDelegate is on the conversion hot path, - // so the common case must stay lock free, but the check and the assignment together are not - // atomic - without the lock two threads racing on first access each build a registry, and - // whatever was registered on the one that loses is silently dropped - synchronized (this) { - answer = delegate; - if (answer == null) { - // ensure the tracker is open so we can discover TypeConverterLoader services - // before creating the registry - this is important because getDelegate() may be - // called during doInit() (e.g. when to() eagerly creates endpoints) which happens - // before doStart() where the tracker is normally opened - ensureTrackerOpen(); - answer = createRegistry(); - delegate = answer; - } - } + register(registry -> registry.setTypeConverterExists(typeConverterExists)); + } + + // 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 + // called during doInit() (e.g. when to() eagerly creates endpoints) which happens + // before doStart() where the tracker is normally opened + ensureTrackerOpen(); + delegate = createRegistry(); } - return answer; + return delegate; } protected DefaultTypeConverter createRegistry() { @@ -299,28 +335,59 @@ public Set getClassLoaders() { 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[] serviceReferences = this.tracker.getServiceReferences(); - if (serviceReferences != null) { - ArrayList> servicesList = - new ArrayList<>(Arrays.asList(serviceReferences)); - // Just make sure we install the high ranking fallback converter at last - Collections.sort(servicesList); - for (ServiceReference 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> servicesList = new ArrayList<>(trackedLoaders.keySet()); + // Just make sure we install the high ranking fallback converter at last + Collections.sort(servicesList); + for (ServiceReference 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 registration : programmaticRegistrations) { + registration.accept(registry); + } + } + + /** + * Applies a registration to the current delegate and remembers it, so that discarding the delegate does not + * discard the registration with it. + */ + private void register(Consumer registration) { + // apply first: a registration the delegate rejects is not one worth replaying. Note getDelegate() may + // build the registry here, which replays the list as it stands - this registration is added after, so + // it cannot be applied twice + registration.accept(getDelegate()); + programmaticRegistrations.add(registration); + } + private class OsgiDefaultTypeConverter extends DefaultTypeConverter { public OsgiDefaultTypeConverter(PackageScanClassResolver resolver, Injector injector, boolean loadTypeConverters, @@ -352,7 +419,7 @@ public Map, TypeConverter> lookup(Class toType) { @Override public void addConverter(TypeConvertible typeConvertible, TypeConverter typeConverter) { - getDelegate().addConverter(typeConvertible, typeConverter); + register(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 3194c1a8e..449bfad94 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,6 +17,8 @@ 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; @@ -56,6 +58,8 @@ public class OsgiTypeConverterTest { @Mock private TypeConverterLoader loader; @Mock + private TypeConverter typeConverter; + @Mock private Bundle bundle; private OsgiTypeConverter osgiTypeConverter; @@ -161,4 +165,102 @@ protected DefaultTypeConverter createRegistry() { "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 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(); + } + } } From 504ef8f0c29db9f0ee9011243da4192e9840b487 Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Thu, 27 Aug 2026 12:12:04 +0200 Subject: [PATCH 3/3] 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) --- .../camel/karaf/core/OsgiTypeConverter.java | 85 ++++++++++----- .../karaf/core/OsgiTypeConverterTest.java | 101 ++++++++++++++++++ 2 files changed, 162 insertions(+), 24 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 5546799b3..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,11 +20,11 @@ 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.concurrent.CopyOnWriteArrayList; import java.util.function.Consumer; import org.apache.camel.CamelContext; @@ -69,11 +69,24 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter, private final Map, TypeConverterLoader> trackedLoaders = new ConcurrentHashMap<>(); /** - * Registrations made through this facade rather than by a {@link TypeConverterLoader}, in the order they were - * made, 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. + * 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. + *

+ * Keyed rather than a plain list, and insertion ordered so replay keeps the original order. The key is what the + * registration is about - 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. + *

+ * Guarded by this instance's monitor, the same one {@link #getDelegate()} takes. */ - private final List> programmaticRegistrations = new CopyOnWriteArrayList<>(); + private final Map> 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; @@ -171,6 +184,12 @@ protected void doStop() throws Exception { 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; } @@ -223,30 +242,33 @@ public T tryConvertTo(Class type, Object value) { @Override public void addTypeConverter(Class toType, Class fromType, TypeConverter typeConverter) { - register(registry -> registry.addTypeConverter(toType, fromType, typeConverter)); + register(new TypeConvertible<>(fromType, toType), + registry -> registry.addTypeConverter(toType, fromType, typeConverter)); } @Override public void addTypeConverters(Object typeConverters) { - register(registry -> registry.addTypeConverters(typeConverters)); + register(typeConverters, registry -> registry.addTypeConverters(typeConverters)); } @Override public void addBulkTypeConverters(BulkTypeConverters bulkTypeConverters) { - register(registry -> registry.addBulkTypeConverters(bulkTypeConverters)); + register(bulkTypeConverters, registry -> registry.addBulkTypeConverters(bulkTypeConverters)); } @Override - public boolean removeTypeConverter(Class toType, Class fromType) { + public synchronized boolean removeTypeConverter(Class toType, Class fromType) { boolean removed = getDelegate().removeTypeConverter(toType, fromType); - // replayed as well, so a rebuild reproduces the sequence rather than resurrecting the converter - programmaticRegistrations.add(registry -> registry.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) { - register(registry -> registry.addFallbackTypeConverter(typeConverter, canPromote)); + register(typeConverter, registry -> registry.addFallbackTypeConverter(typeConverter, canPromote)); } @Override @@ -256,7 +278,7 @@ public TypeConverter lookup(Class toType, Class fromType) { @Override public void setInjector(Injector injector) { - register(registry -> registry.setInjector(injector)); + register(INJECTOR_KEY, registry -> registry.setInjector(injector)); } @Override @@ -281,7 +303,8 @@ public LoggingLevel getTypeConverterExistsLoggingLevel() { @Override public void setTypeConverterExistsLoggingLevel(LoggingLevel loggingLevel) { - register(registry -> registry.setTypeConverterExistsLoggingLevel(loggingLevel)); + register(TYPE_CONVERTER_EXISTS_LOGGING_LEVEL_KEY, + registry -> registry.setTypeConverterExistsLoggingLevel(loggingLevel)); } @Override @@ -291,7 +314,7 @@ public TypeConverterExists getTypeConverterExists() { @Override public void setTypeConverterExists(TypeConverterExists typeConverterExists) { - register(registry -> registry.setTypeConverterExists(typeConverterExists)); + register(TYPE_CONVERTER_EXISTS_KEY, registry -> registry.setTypeConverterExists(typeConverterExists)); } // fully synchronized rather than double checked: the delegate is not immutable after publication - @@ -371,21 +394,35 @@ private void replayProgrammaticRegistrations(DefaultTypeConverter registry) { } LOG.debug("Replaying {} programmatic registration(s) onto the rebuilt type converter registry", programmaticRegistrations.size()); - for (Consumer registration : programmaticRegistrations) { + for (Consumer registration : programmaticRegistrations.values()) { registration.accept(registry); } } /** - * Applies a registration to the current delegate and remembers it, so that discarding the delegate does not - * discard the registration with it. + * 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. + *

+ * 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. */ - private void register(Consumer registration) { - // apply first: a registration the delegate rejects is not one worth replaying. Note getDelegate() may - // build the registry here, which replays the list as it stands - this registration is added after, so - // it cannot be applied twice + synchronized int programmaticRegistrationCount() { + return programmaticRegistrations.size(); + } + + private synchronized void register(Object key, Consumer 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.add(registration); + programmaticRegistrations.put(key, registration); } private class OsgiDefaultTypeConverter extends DefaultTypeConverter { @@ -419,7 +456,7 @@ public Map, TypeConverter> lookup(Class toType) { @Override public void addConverter(TypeConvertible typeConvertible, TypeConverter typeConverter) { - register(registry -> registry.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 449bfad94..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 @@ -27,6 +27,7 @@ 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; @@ -263,4 +264,104 @@ void addingServiceMustNotHoldTheInstanceMonitor() throws Exception { 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 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(); + } + } }