SAI-5884: hook lucene Unloader into Solr container lifecycle

.github/workflows/solr-test.yml

@@ -36,4 +36,4 @@ jobs:
     - name: Initialize gradle settings
       run: ./gradlew localSettings
     - name: Test Solr
-      run: ./gradlew test -Ptests.directory=org.apache.solr.storage.TeeDirectory
+      run: ./gradlew test -Ptests.directory=org.apache.solr.storage.TeeDirectory -Plucene.unload.ttl=1 -Plucene.unload.initial=0

gradle/testing/defaults-tests.gradle

@@ -31,6 +31,10 @@ allprojects {
       // completes.
       // [propName: 'tests.foo', value: "bar", description: "Sets foo in tests."],
       testOptions = [
+          [propName: 'lucene.unload.ttl', value: '60m', description: "Time since last use at which a resource becomes eligible for unloading"],
+          [propName: 'lucene.unload.initial', value: '1m', description: "Extra time allotted for first resource use after load/reload"],
+          [propName: 'lucene.unload.disable', value: 'false', description: "Disables resource unloading"],
+          [propName: 'lucene.unload.trackReloads', value: 'false', description: "tracks reloads via logs and metrics"],
           // asserts, debug output.
           [propName: 'tests.verbose', value: false, description: "Enables verbose mode (emits full test outputs immediately)."],
           [propName: 'tests.workDir',

solr/core/src/java/org/apache/solr/core/CachingDirectoryFactory.java

@@ -19,6 +19,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.lang.invoke.MethodHandles;
+import java.lang.ref.WeakReference;
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -31,10 +32,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import org.apache.lucene.index.Unloader;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.LockFactory;
+import org.apache.lucene.store.UnloaderCoordinationPoint;
 import org.apache.lucene.util.IOUtils;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.SolrException.ErrorCode;
@@ -408,6 +412,7 @@ public final Directory get(String path, DirContext dirContext, String rawLockTyp
 
       if (directory == null) {
         directory = create(fullPath, createLockFactory(rawLockType), dirContext);
+        UnloaderCoordinationPoint.setUnloadHelperSupplier(directory, unloadHelperSupplier);
         assert ObjectReleaseTracker.track(directory);
         boolean success = false;
         try {
@@ -430,6 +435,19 @@ public final Directory get(String path, DirContext dirContext, String rawLockTyp
     }
   }
 
+  private volatile WeakReference<CoreContainer> cc;
+  private volatile Supplier<Unloader.UnloadHelper> unloadHelperSupplier;
+
+  @Override
+  public void initCoreContainer(CoreContainer cc) {
+    super.initCoreContainer(cc);
+    // no purpose setting up unloading on nodes that don't use it
+    if (cc == null || cc.nodeRoles.getRoleMode(NodeRoles.Role.DATA).equals(NodeRoles.MODE_OFF)) {
+      return;
+    }
+    this.cc = new WeakReference<>(cc);
+  }
+
   /*
    * (non-Javadoc)
    *
@@ -454,7 +472,23 @@ public void incRef(Directory directory) {
   }
 
   @Override
+  @SuppressWarnings("unchecked")
   public void init(NamedList<?> args) {
+    CoreContainer cc;
+    String duff = (String) args.get("disableUnloadForField");
+    if (duff != null && duff.isEmpty()) {
+      duff = null;
+    }
+    String duffF = duff;
+    int unloadMinSegSize = args.toSolrParams().getInt("unloadMinSegSize", 2_000);
+    if (this.cc != null && (cc = this.cc.get()) != null) {
+      unloadHelperSupplier =
+          (Supplier<Unloader.UnloadHelper>)
+              cc.getObjectCache()
+                  .computeIfAbsent(
+                      "nodeLevelUnloadMetrics",
+                      (k) -> new UnloadHelper<>(cc, duffF, unloadMinSegSize));
+    }
     maxWriteMBPerSecFlush = (Double) args.get("maxWriteMBPerSecFlush");
     maxWriteMBPerSecMerge = (Double) args.get("maxWriteMBPerSecMerge");
     maxWriteMBPerSecRead = (Double) args.get("maxWriteMBPerSecRead");

solr/core/src/java/org/apache/solr/core/CoreContainer.java

@@ -54,6 +54,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -189,6 +191,13 @@ public Executor getCollectorExecutor() {
     return collectorExecutor;
   }
 
+  private final ScheduledExecutorService unloaderExecutor =
+      Executors.newSingleThreadScheduledExecutor(new SolrNamedThreadFactory("unloaderExecutor"));
+
+  public ScheduledExecutorService getUnloaderExecutor() {
+    return unloaderExecutor;
+  }
+
   public static class CoreLoadFailure {
 
     public final CoreDescriptor cd;
@@ -1279,6 +1288,7 @@ public boolean isShutDown() {
   /** Close / shut down. Only called by {@link org.apache.solr.servlet.CoreContainerProvider}. */
   public void shutdown() {
 
+    unloaderExecutor.shutdownNow();
     ZkController zkController = getZkController();
     if (zkController != null) {
       if (distributedCollectionCommandRunner.isPresent()) {
@@ -1296,6 +1306,7 @@ public void shutdown() {
 
     ExecutorUtil.shutdownAndAwaitTermination(coreContainerAsyncTaskExecutor);
     ExecutorUtil.shutdownAndAwaitTermination(collectorExecutor);
+    ExecutorUtil.awaitTermination(unloaderExecutor);
     ExecutorService customThreadPool =
         ExecutorUtil.newMDCAwareCachedThreadPool(new SolrNamedThreadFactory("closeThreadPool"));