Deprecation of VFS Class Loader items, new Context class loader management

core/src/main/java/org/apache/accumulo/core/classloader/ClassLoaderUtil.java

@@ -0,0 +1,34 @@
+/*
+ * 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.accumulo.core.classloader;
+
+import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
+
+public class ClassLoaderUtil {
+
+  public static <U> Class<? extends U> loadClass(String contextName, String className,
+      Class<U> extension) throws ClassNotFoundException {
+    if (contextName != null && !contextName.isEmpty())
+      return ContextClassLoaders.getContextClassLoaderFactory().getClassLoader(contextName)
+          .loadClass(className).asSubclass(extension);
+    else
+      return AccumuloVFSClassLoader.loadClass(className, extension);
+
+  }
+}

core/src/main/java/org/apache/accumulo/core/classloader/ContextClassLoaders.java

@@ -0,0 +1,88 @@
+/*
+ * 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.accumulo.core.classloader;
+
+import java.lang.reflect.InvocationTargetException;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.spi.common.ContextClassLoaderFactory;
+import org.apache.accumulo.core.util.ConfigurationImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ContextClassLoaders {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ContextClassLoaders.class);
+
+  private static ContextClassLoaderFactory FACTORY;
+  private static AccumuloConfiguration CONF;
+
+  /**
+   * Initialize the ContextClassLoaderFactory
+   *
+   * @param conf
+   *          AccumuloConfiguration object
+   */
+  public static synchronized void initialize(AccumuloConfiguration conf) throws Exception {
+    if (null == CONF) {
+      CONF = conf;
+      LOG.info("Creating ContextClassLoaderFactory");
+      var factoryName = CONF.get(Property.GENERAL_CONTEXT_CLASSLOADER_FACTORY.toString());
+      if (null == factoryName || factoryName.isEmpty()) {
+        LOG.info("No ClassLoaderFactory specified");
+        return;
+      }
+      try {
+        var factoryClass = Class.forName(factoryName);
+        if (ContextClassLoaderFactory.class.isAssignableFrom(factoryClass)) {
+          LOG.info("Creating ContextClassLoaderFactory: {}", factoryName);
+          FACTORY = ((Class<? extends ContextClassLoaderFactory>) factoryClass)
+              .getDeclaredConstructor().newInstance();
+          FACTORY.initialize(new ConfigurationImpl(CONF));
+        } else {
+          throw new RuntimeException(factoryName + " does not implement ContextClassLoaderFactory");
+        }
+      } catch (ClassNotFoundException | InstantiationException | IllegalAccessException
+          | IllegalArgumentException | InvocationTargetException | NoSuchMethodException
+          | SecurityException e) {
+        LOG.error(
+            "Unable to load and initialize class: {}. Ensure that the jar containing the ContextClassLoaderFactory is on the classpath",
+            factoryName);
+        throw e;
+      }
+    } else {
+      LOG.debug("ContextClassLoaderFactory already initialized.");
+    }
+  }
+
+  /**
+   * Get the ContextClassLoaderFactory
+   *
+   * @return the configured context classloader factory
+   */
+  public static ContextClassLoaderFactory getContextClassLoaderFactory() {
+    return FACTORY;
+  }
+
+  public static void resetForTests() {
+    CONF = null;
+  }
+
+}

core/src/main/java/org/apache/accumulo/core/classloader/LegacyVFSContextClassLoaderFactory.java

@@ -0,0 +1,94 @@
+/*
+ * 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.accumulo.core.classloader;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import org.apache.accumulo.core.client.PluginEnvironment.Configuration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.spi.common.ContextClassLoaderFactory;
+import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
+import org.apache.accumulo.start.classloader.vfs.ContextManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Deprecated(since = "2.1.0", forRemoval = true)
+public class LegacyVFSContextClassLoaderFactory implements ContextClassLoaderFactory {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(LegacyVFSContextClassLoaderFactory.class);
+
+  public void initialize(Configuration contextProperties) {
+    try {
+      AccumuloVFSClassLoader.getContextManager()
+          .setContextConfig(new ContextManager.DefaultContextsConfig() {
+            @Override
+            public Map<String,String> getVfsContextClasspathProperties() {
+              return contextProperties
+                  .getWithPrefix(Property.VFS_CONTEXT_CLASSPATH_PROPERTY.getKey());
+            }
+          });
+      LOG.debug("ContextManager configuration set");
+      new Timer("LegacyVFSContextClassLoaderFactory-cleanup", true)
+          .scheduleAtFixedRate(new TimerTask() {
+            @Override
+            public void run() {
+              try {
+                if (LOG.isTraceEnabled()) {
+                  LOG.trace("LegacyVFSContextClassLoaderFactory-cleanup thread, properties: {}",
+                      contextProperties);
+                }
+                Set<String> configuredContexts = new HashSet<>();
+                contextProperties.getWithPrefix(Property.VFS_CONTEXT_CLASSPATH_PROPERTY.getKey())
+                    .forEach((k, v) -> {
+                      configuredContexts.add(
+                          k.substring(Property.VFS_CONTEXT_CLASSPATH_PROPERTY.getKey().length()));
+                    });
+                LOG.trace("LegacyVFSContextClassLoaderFactory-cleanup thread, contexts in use: {}",
+                    configuredContexts);
+                AccumuloVFSClassLoader.getContextManager().removeUnusedContexts(configuredContexts);
+              } catch (IOException e) {
+                LOG.warn("{}", e.getMessage(), e);
+              }
+            }
+          }, 60000, 60000);
+      LOG.debug("Context cleanup timer started at 60s intervals");
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+
+  }
+
+  @Override
+  public ClassLoader getClassLoader(String contextName) throws IllegalArgumentException {
+    try {
+      return AccumuloVFSClassLoader.getContextManager().getClassLoader(contextName);
+    } catch (IOException e) {
+      throw new UncheckedIOException(
+          "Error getting context class loader for context: " + contextName, e);
+    }
+  }
+
+}

core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java

@@ -24,7 +24,7 @@
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
+import org.apache.accumulo.core.classloader.ClassLoaderUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -200,13 +200,7 @@ public static <T> T getClassInstance(String context, String clazzName, Class<T>
       throws IOException, ReflectiveOperationException {
     T instance;
 
-    Class<? extends T> clazz;
-    if (context != null && !context.isEmpty()) {
-      clazz = AccumuloVFSClassLoader.getContextManager().loadClass(context, clazzName, base);
-    } else {
-      clazz = AccumuloVFSClassLoader.loadClass(clazzName, base);
-    }
-
+    Class<? extends T> clazz = ClassLoaderUtil.loadClass(context, clazzName, base);
     instance = clazz.getDeclaredConstructor().newInstance();
     if (loaded.put(clazzName, clazz) != clazz)
       log.debug("Loaded class : {}", clazzName);

core/src/main/java/org/apache/accumulo/core/conf/IterConfigUtil.java

@@ -31,6 +31,7 @@
 import java.util.Set;
 import java.util.TreeMap;
 
+import org.apache.accumulo.core.classloader.ClassLoaderUtil;
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.constraints.DefaultKeySizeConstraint;
 import org.apache.accumulo.core.data.Key;
@@ -40,7 +41,6 @@
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.iterators.user.VersioningIterator;
-import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -231,20 +231,12 @@ public static SortedKeyValueIterator<Key,Value> loadIterators(
   @SuppressWarnings("unchecked")
   private static Class<SortedKeyValueIterator<Key,Value>> loadClass(boolean useAccumuloClassLoader,
       String context, IterInfo iterInfo) throws ClassNotFoundException, IOException {
-    Class<SortedKeyValueIterator<Key,Value>> clazz;
+    Class<SortedKeyValueIterator<Key,Value>> clazz = null;
     if (useAccumuloClassLoader) {
-      if (context != null && !context.equals("")) {
-        clazz =
-            (Class<SortedKeyValueIterator<Key,Value>>) AccumuloVFSClassLoader.getContextManager()
-                .loadClass(context, iterInfo.className, SortedKeyValueIterator.class);
-        log.trace("Iterator class {} loaded from context {}, classloader: {}", iterInfo.className,
-            context, clazz.getClassLoader());
-      } else {
-        clazz = (Class<SortedKeyValueIterator<Key,Value>>) AccumuloVFSClassLoader
-            .loadClass(iterInfo.className, SortedKeyValueIterator.class);
-        log.trace("Iterator class {} loaded from AccumuloVFSClassLoader: {}", iterInfo.className,
-            clazz.getClassLoader());
-      }
+      clazz = (Class<SortedKeyValueIterator<Key,Value>>) ClassLoaderUtil.loadClass(context,
+          iterInfo.className, SortedKeyValueIterator.class);
+      log.trace("Iterator class {} loaded from context {}, classloader: {}", iterInfo.className,
+          context, clazz.getClassLoader());
     } else {
       clazz = (Class<SortedKeyValueIterator<Key,Value>>) Class.forName(iterInfo.className)
           .asSubclass(SortedKeyValueIterator.class);

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -26,6 +26,7 @@
 import java.util.Map.Entry;
 
 import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.classloader.LegacyVFSContextClassLoaderFactory;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
 import org.apache.accumulo.core.constraints.NoDeleteConstraint;
 import org.apache.accumulo.core.file.rfile.RFile;
@@ -196,6 +197,9 @@ public enum Property {
           + "Additionally, this property no longer does property interpolation of environment "
           + "variables, such as '$ACCUMULO_HOME'. Use commons-configuration syntax,"
           + "'${env:ACCUMULO_HOME}' instead."),
+  GENERAL_CONTEXT_CLASSLOADER_FACTORY("general.context.class.loader.factory",
+      LegacyVFSContextClassLoaderFactory.class.getName(), PropertyType.STRING,
+      "Name of classloader factory to be used to create classloaders for named contexts."),
   GENERAL_RPC_TIMEOUT("general.rpc.timeout", "120s", PropertyType.TIMEDURATION,
       "Time to wait on I/O for simple, short RPC calls"),
   @Experimental
@@ -923,10 +927,12 @@ public enum Property {
 
   // this property shouldn't be used directly; it exists solely to document the default value
   // defined by its use in AccumuloVFSClassLoader when generating the property documentation
+  @Deprecated(since = "2.1.0", forRemoval = true)
   VFS_CLASSLOADER_SYSTEM_CLASSPATH_PROPERTY(
       AccumuloVFSClassLoader.VFS_CLASSLOADER_SYSTEM_CLASSPATH_PROPERTY, "", PropertyType.STRING,
       "Configuration for a system level vfs classloader. Accumulo jar can be"
           + " configured here and loaded out of HDFS."),
+  @Deprecated(since = "2.1.0", forRemoval = true)
   VFS_CONTEXT_CLASSPATH_PROPERTY(AccumuloVFSClassLoader.VFS_CONTEXT_CLASSPATH_PROPERTY, null,
       PropertyType.PREFIX,
       "Properties in this category are define a classpath. These properties"
@@ -941,6 +947,7 @@ public enum Property {
 
   // this property shouldn't be used directly; it exists solely to document the default value
   // defined by its use in AccumuloVFSClassLoader when generating the property documentation
+  @Deprecated(since = "2.1.0", forRemoval = true)
   VFS_CLASSLOADER_CACHE_DIR(AccumuloVFSClassLoader.VFS_CACHE_DIR, "${java.io.tmpdir}",
       PropertyType.ABSOLUTEPATH,
       "The base directory to use for the vfs cache. The actual cached files will be located"

core/src/main/java/org/apache/accumulo/core/file/BloomFilterLayer.java

@@ -37,6 +37,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.accumulo.core.bloomfilter.DynamicBloomFilter;
+import org.apache.accumulo.core.classloader.ClassLoaderUtil;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.ConfigurationCopy;
 import org.apache.accumulo.core.conf.DefaultConfiguration;
@@ -54,7 +55,6 @@
 import org.apache.accumulo.core.sample.impl.SamplerConfigurationImpl;
 import org.apache.accumulo.core.util.NamingThreadFactory;
 import org.apache.accumulo.fate.util.LoggingRunnable;
-import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.io.Text;
@@ -127,11 +127,8 @@ private synchronized void initBloomFilter(AccumuloConfiguration acuconf,
         Class<? extends KeyFunctor> clazz;
         if (!useAccumuloStart)
           clazz = Writer.class.getClassLoader().loadClass(classname).asSubclass(KeyFunctor.class);
-        else if (context != null && !context.equals(""))
-          clazz = AccumuloVFSClassLoader.getContextManager().loadClass(context, classname,
-              KeyFunctor.class);
         else
-          clazz = AccumuloVFSClassLoader.loadClass(classname, KeyFunctor.class);
+          clazz = ClassLoaderUtil.loadClass(context, classname, KeyFunctor.class);
 
         transformer = clazz.getDeclaredConstructor().newInstance();
 
@@ -238,12 +235,8 @@ static class BloomFilterLoader {
            */
           ClassName = in.readUTF();
 
-          Class<? extends KeyFunctor> clazz;
-          if (context != null && !context.equals(""))
-            clazz = AccumuloVFSClassLoader.getContextManager().loadClass(context, ClassName,
-                KeyFunctor.class);
-          else
-            clazz = AccumuloVFSClassLoader.loadClass(ClassName, KeyFunctor.class);
+          Class<? extends KeyFunctor> clazz =
+              ClassLoaderUtil.loadClass(context, ClassName, KeyFunctor.class);
           transformer = clazz.getDeclaredConstructor().newInstance();
 
           /**

core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/BlockCacheManagerFactory.java

@@ -18,10 +18,10 @@
  */
 package org.apache.accumulo.core.file.blockfile.cache.impl;
 
+import org.apache.accumulo.core.classloader.ClassLoaderUtil;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.spi.cache.BlockCacheManager;
-import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,7 +43,7 @@ public static synchronized BlockCacheManager getInstance(AccumuloConfiguration c
       throws Exception {
     String impl = conf.get(Property.TSERV_CACHE_MANAGER_IMPL);
     Class<? extends BlockCacheManager> clazz =
-        AccumuloVFSClassLoader.loadClass(impl, BlockCacheManager.class);
+        ClassLoaderUtil.loadClass(null, impl, BlockCacheManager.class);
     LOG.info("Created new block cache manager of type: {}", clazz.getSimpleName());
     return clazz.getDeclaredConstructor().newInstance();
   }