FELIX-4385: Applied the patch fixing the NPE when configuration is deleted at the same time
the corresponding ManagedService service is unregistered from the OSGi registry.
Also added the corresponding integration test.
git-svn-id: https://svn.apache.org/repos/asf/felix/trunk@1563946 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/configadmin/src/main/java/org/apache/felix/cm/impl/ConfigurationManager.java b/configadmin/src/main/java/org/apache/felix/cm/impl/ConfigurationManager.java
index 0bb8b7f..12dd0de 100644
--- a/configadmin/src/main/java/org/apache/felix/cm/impl/ConfigurationManager.java
+++ b/configadmin/src/main/java/org/apache/felix/cm/impl/ConfigurationManager.java
@@ -1663,6 +1663,9 @@
try
{
final String configPidString = this.getHelper().getServicePid( sr, this.config.getPid() );
+ if (configPidString == null) {
+ return false; // The managed service is not registered anymore in the OSGi service registry.
+ }
final ConfigurationImpl rc = getTargetedConfiguration( configPidString, sr );
if ( rc != null )
{
@@ -1723,7 +1726,15 @@
if ( !srList.isEmpty() )
{
// optionally bind dynamically to the first service
- config.tryBindLocation( srList.get( 0 ).getBundle().getLocation() );
+ Bundle bundle = srList.get(0).getBundle();
+ if (bundle == null) {
+ log( LogService.LOG_DEBUG,
+ "Service {0} seems to be unregistered concurrently (not providing configuration)",
+ new Object[]
+ { ConfigurationManager.toString( srList.get(0) ) } );
+ return;
+ }
+ config.tryBindLocation( bundle.getLocation() );
final String configBundleLocation = config.getBundleLocation();
diff --git a/configadmin/src/test/java/org/apache/felix/cm/integration/FELIX4385_StressTest.java b/configadmin/src/test/java/org/apache/felix/cm/integration/FELIX4385_StressTest.java
new file mode 100644
index 0000000..659ce88
--- /dev/null
+++ b/configadmin/src/test/java/org/apache/felix/cm/integration/FELIX4385_StressTest.java
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.felix.cm.integration;
+
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.Assert;
+import junit.framework.TestCase;
+
+import org.apache.felix.cm.integration.helper.Log;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.junit.JUnit4TestRunner;
+import org.osgi.framework.Constants;
+import org.osgi.framework.ServiceRegistration;
+import org.osgi.service.cm.Configuration;
+import org.osgi.service.cm.ConfigurationAdmin;
+import org.osgi.service.cm.ConfigurationException;
+import org.osgi.service.cm.ManagedService;
+
+/**
+ * The <code>FELIX4385_StressTest</code> class tests the issue related to concurrency between configuration
+ * creation/update/removal and ManagedService registration/unregistration.
+ * The test performs some loops, each one is then executing the following scenario:
+ * Some ManagedServices are concurrently registered in the OSGi registry using an Executor, and for each
+ * managed service, we create a Configuration.
+ * We then wait until every managed services have been updated with a non null configuration. Care is taken when a
+ * ManagedService is called with an initial update(null) callback, because when a configuration is created the very first
+ * time, an empty configuration is delivered to the corresponding managed service until the configuration is really updated.
+ * Once all managed services have been updated, we then concurrently unregister the managed services, and we also
+ * delete every created configurations. We don't use an executor when deleting configuration because the configuration
+ * removal is already asynchronous.
+ *
+ * <p>
+ * @see <a href="https://issues.apache.org/jira/browse/FELIX-4385">FELIX-4385</a>
+ */
+@RunWith(JUnit4TestRunner.class)
+public class FELIX4385_StressTest extends ConfigurationTestBase
+{
+ final static int MAXWAIT = 10000;
+ final static int MANAGED_SERVICES = 3;
+ volatile ExecutorService executor;
+
+ @Test
+ public void test_ConcurrentManagedServicesWithConcurrentConfigurations()
+ {
+ final Log log = new Log(bundleContext);
+ log.info("starting test_ConcurrentManagedServicesWithConcurrentConfigurations");
+ // Use at least 10 parallel threads, or take all available processors if the running host contains more than 10 processors.
+ int parallelism = Math.max(10, Runtime.getRuntime().availableProcessors());
+ final ConfigurationAdmin ca = getConfigurationAdmin();
+ final ExecutorService executor = Executors.newFixedThreadPool(parallelism);
+ try
+ {
+ int pidCounter = 0;
+
+ long timeStamp = System.currentTimeMillis();
+ for (int loop = 0; loop < 1000; loop++)
+ {
+ log.debug("loop#%d -------------------------", (loop + 1));
+
+ final CountDownLatch managedServiceUpdated = new CountDownLatch(MANAGED_SERVICES);
+ final CountDownLatch managedServiceUnregistered = new CountDownLatch(MANAGED_SERVICES);
+
+ // Create some ManagedServices concurrently
+ log.info("registering aspects concurrently");
+ final CopyOnWriteArrayList<ServiceRegistration> managedServices = new CopyOnWriteArrayList<ServiceRegistration>();
+ final CopyOnWriteArrayList<Configuration> confs = new CopyOnWriteArrayList<Configuration>();
+
+ for (int i = 0; i < MANAGED_SERVICES; i++)
+ {
+ final String pid = "pid." + i + "-" + (pidCounter++);
+ executor.execute(new Runnable()
+ {
+ public void run()
+ {
+ Hashtable props = new Hashtable();
+ props.put(Constants.SERVICE_PID, pid);
+
+ ServiceRegistration sr = bundleContext.registerService(
+ ManagedService.class.getName(),
+ new TestManagedService(managedServiceUpdated), props);
+ managedServices.add(sr);
+ try
+ {
+ Configuration c = ca.getConfiguration(pid, null);
+ c.update(new Hashtable()
+ {
+ {
+ put("foo", "bar");
+ }
+ });
+ confs.add(c);
+ }
+ catch (IOException e)
+ {
+ log.error("could not create pid %s", e, pid);
+ return;
+ }
+ }
+ });
+ }
+
+ if (!managedServiceUpdated.await(MAXWAIT, TimeUnit.MILLISECONDS))
+ {
+ TestCase.fail("Detected errors logged during concurrent test");
+ break;
+ }
+ log.info("all managed services updated");
+
+ // Unregister managed services concurrently
+ log.info("unregistering services concurrently");
+ for (final ServiceRegistration sr : managedServices)
+ {
+ executor.execute(new Runnable()
+ {
+ public void run()
+ {
+ sr.unregister();
+ managedServiceUnregistered.countDown();
+ }
+ });
+ }
+
+ // Unregister configuration concurrently
+ log.info("unregistering configuration concurrently");
+ for (final Configuration c : confs)
+ {
+ c.delete();
+ }
+
+ // Wait until managed services have been unregistered
+ if (!managedServiceUnregistered.await(MAXWAIT, TimeUnit.MILLISECONDS))
+ {
+ TestCase.fail("Managed Servives could not be unregistered timely");
+ break;
+ }
+
+ if (log.errorsLogged())
+ {
+ TestCase.fail("Detected errors logged during concurrent test");
+ break;
+ }
+
+ log.info("finished one test loop");
+ if ((loop + 1) % 100 == 0)
+ {
+ long duration = System.currentTimeMillis() - timeStamp;
+ System.out.println(String.format("Performed %d tests in %d ms.", (loop + 1), duration));
+ timeStamp = System.currentTimeMillis();
+ }
+ }
+ }
+
+ catch (Throwable t)
+ {
+ Assert.fail("Test failed: " + t.getMessage());
+ }
+
+ finally
+ {
+ shutdown(executor);
+ log.close();
+ }
+ }
+
+ void shutdown(ExecutorService exec)
+ {
+ exec.shutdown();
+ try
+ {
+ exec.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ catch (InterruptedException e)
+ {
+ }
+ }
+
+ /**
+ * One ManagedService concurrently registered in the OSGI registry.
+ * We count down a latch once we have been updated with our configuration.
+ */
+ public class TestManagedService implements ManagedService
+ {
+ private final CountDownLatch latch;
+ private Dictionary<String, ?> props;
+
+ TestManagedService(CountDownLatch latch)
+ {
+ this.latch = latch;
+ }
+
+ public synchronized void updated(Dictionary<String, ?> properties) throws ConfigurationException
+ {
+ if (this.props == null && properties == null)
+ {
+ // GetConfiguration has been called, but configuration have not yet been delivered.
+ return;
+ }
+ this.props = properties;
+ latch.countDown();
+ }
+ }
+}
diff --git a/configadmin/src/test/java/org/apache/felix/cm/integration/helper/Log.java b/configadmin/src/test/java/org/apache/felix/cm/integration/helper/Log.java
new file mode 100644
index 0000000..d30da00
--- /dev/null
+++ b/configadmin/src/test/java/org/apache/felix/cm/integration/helper/Log.java
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.felix.cm.integration.helper;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.FrameworkEvent;
+import org.osgi.framework.FrameworkListener;
+import org.osgi.framework.ServiceReference;
+import org.osgi.framework.ServiceRegistration;
+import org.osgi.service.log.LogService;
+
+/**
+ * OSGi log service which logs messages to standard output.
+ * This class can also be used to detect if the ConfigurationAdmin service has logged
+ * some warnings during a stress integration test.
+ */
+public class Log implements LogService, FrameworkListener
+{
+ // Default OSGI log service level logged to standard output.
+ private final static int LOG_LEVEL = LogService.LOG_WARNING;
+
+ // Flag used to check if some errors have been logged during the execution of a given test.
+ private volatile boolean m_errorsLogged;
+
+ // We implement OSGI log service.
+ protected ServiceRegistration logService;
+
+ // Bundle context used to register our log listener
+ private BundleContext ctx;
+
+ /**
+ * Default constructor.
+ * @Param ctx the Bundle Context used to register this log service. The {@link #close} must
+ * be called when the logger is not used anymore.
+ */
+ public Log(BundleContext ctx)
+ {
+ this.ctx = ctx;
+ logService = ctx.registerService(LogService.class.getName(), this, null);
+ ctx.addFrameworkListener(this);
+ }
+
+ /**
+ * Unregister our log listener
+ */
+ public void close()
+ {
+ logService.unregister();
+ ctx.removeFrameworkListener(this);
+ }
+
+ public void log(int level, String message)
+ {
+ checkError(level, null);
+ if (LOG_LEVEL >= level)
+ {
+ System.out.println(getLevel(level) + " - " + Thread.currentThread().getName() + " : " + message);
+ }
+ }
+
+ public void log(int level, String message, Throwable exception)
+ {
+ checkError(level, exception);
+ if (LOG_LEVEL >= level)
+ {
+ StringBuilder sb = new StringBuilder();
+ sb.append(getLevel(level) + " - " + Thread.currentThread().getName() + " : ");
+ sb.append(message);
+ parse(sb, exception);
+ System.out.println(sb.toString());
+ }
+ }
+
+ public void log(ServiceReference sr, int level, String message)
+ {
+ checkError(level, null);
+ if (LOG_LEVEL >= level)
+ {
+ StringBuilder sb = new StringBuilder();
+ sb.append(getLevel(level) + " - " + Thread.currentThread().getName() + " : ");
+ sb.append(message);
+ System.out.println(sb.toString());
+ }
+ }
+
+ public void log(ServiceReference sr, int level, String message, Throwable exception)
+ {
+ checkError(level, exception);
+ if (LOG_LEVEL >= level)
+ {
+ StringBuilder sb = new StringBuilder();
+ sb.append(getLevel(level) + " - " + Thread.currentThread().getName() + " : ");
+ sb.append(message);
+ parse(sb, exception);
+ System.out.println(sb.toString());
+ }
+ }
+
+ public boolean errorsLogged()
+ {
+ return m_errorsLogged;
+ }
+
+ private void parse(StringBuilder sb, Throwable t)
+ {
+ if (t != null)
+ {
+ sb.append(" - ");
+ StringWriter buffer = new StringWriter();
+ PrintWriter pw = new PrintWriter(buffer);
+ t.printStackTrace(pw);
+ sb.append(buffer.toString());
+ m_errorsLogged = true;
+ }
+ }
+
+ private String getLevel(int level)
+ {
+ switch (level)
+ {
+ case LogService.LOG_DEBUG:
+ return "DEBUG";
+ case LogService.LOG_ERROR:
+ return "ERROR";
+ case LogService.LOG_INFO:
+ return "INFO";
+ case LogService.LOG_WARNING:
+ return "WARN";
+ default:
+ return "";
+ }
+ }
+
+ private void checkError(int level, Throwable exception)
+ {
+ if (level <= LOG_ERROR)
+ {
+ m_errorsLogged = true;
+ }
+ if (exception != null)
+ {
+ m_errorsLogged = true;
+ }
+ }
+
+ public void frameworkEvent(FrameworkEvent event)
+ {
+ int eventType = event.getType();
+ String msg = getFrameworkEventMessage(eventType);
+ int level = (eventType == FrameworkEvent.ERROR) ? LOG_ERROR : LOG_WARNING;
+ if (msg != null)
+ {
+ log(level, msg, event.getThrowable());
+ }
+ else
+ {
+ log(level, "Unknown fwk event: " + event);
+ }
+ }
+
+ private String getFrameworkEventMessage(int event)
+ {
+ switch (event)
+ {
+ case FrameworkEvent.ERROR:
+ return "FrameworkEvent: ERROR";
+ case FrameworkEvent.INFO:
+ return "FrameworkEvent INFO";
+ case FrameworkEvent.PACKAGES_REFRESHED:
+ return "FrameworkEvent: PACKAGE REFRESHED";
+ case FrameworkEvent.STARTED:
+ return "FrameworkEvent: STARTED";
+ case FrameworkEvent.STARTLEVEL_CHANGED:
+ return "FrameworkEvent: STARTLEVEL CHANGED";
+ case FrameworkEvent.WARNING:
+ return "FrameworkEvent: WARNING";
+ default:
+ return null;
+ }
+ }
+
+ public void warn(String msg, Object... params)
+ {
+ if (LOG_LEVEL >= LogService.LOG_WARNING)
+ {
+ log(LogService.LOG_WARNING, params.length > 0 ? String.format(msg, params) : msg);
+ }
+ }
+
+ public void info(String msg, Object... params)
+ {
+ if (LOG_LEVEL >= LogService.LOG_INFO)
+ {
+ log(LogService.LOG_INFO, params.length > 0 ? String.format(msg, params) : msg);
+ }
+ }
+
+ public void debug(String msg, Object... params)
+ {
+ if (LOG_LEVEL >= LogService.LOG_DEBUG)
+ {
+ log(LogService.LOG_DEBUG, params.length > 0 ? String.format(msg, params) : msg);
+ }
+ }
+
+ public void error(String msg, Object... params)
+ {
+ log(LogService.LOG_ERROR, params.length > 0 ? String.format(msg, params) : msg);
+ }
+
+ public void error(String msg, Throwable err, Object... params)
+ {
+ log(LogService.LOG_ERROR, params.length > 0 ? String.format(msg, params) : msg, err);
+ }
+
+ public void error(Throwable err)
+ {
+ log(LogService.LOG_ERROR, "error", err);
+ }
+}