Force execution of SearchService.Reaper (#106555)

If the search threadpool fills up then we may reject execution of
`SearchService.Reaper` which means it stops retrying. We must instead
force its execution so that it keeps on going.

With #106542, closes #106543
Backport of #106544 to 7.17

docs/changelog/106544.yaml

@@ -0,0 +1,6 @@
+pr: 106544
+summary: Force execution of `SearchService.Reaper`
+area: Search
+type: bug
+issues:
+ - 106543

server/src/main/java/org/elasticsearch/search/SearchService.java

@@ -39,7 +39,9 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.TimeValue;
@@ -1434,9 +1436,9 @@ public ResponseCollectorService getResponseCollectorService() {
         return this.responseCollectorService;
     }
 
-    class Reaper implements Runnable {
+    class Reaper extends AbstractRunnable {
         @Override
-        public void run() {
+        protected void doRun() {
             assert Transports.assertNotTransportThread("closing contexts may do IO, e.g. deleting dangling files")
                 && ThreadPool.assertNotScheduleThread("closing contexts may do IO, e.g. deleting dangling files");
             for (ReaderContext context : activeReaders.values()) {
@@ -1446,6 +1448,27 @@ public void run() {
                 }
             }
         }
+
+        @Override
+        public void onFailure(Exception e) {
+            logger.error("unexpected error when freeing search contexts", e);
+            assert false : e;
+        }
+
+        @Override
+        public void onRejection(Exception e) {
+            if (e instanceof EsRejectedExecutionException && ((EsRejectedExecutionException) e).isExecutorShutdown()) {
+                logger.debug("rejected execution when freeing search contexts");
+            } else {
+                onFailure(e);
+            }
+        }
+
+        @Override
+        public boolean isForceExecution() {
+            // mustn't reject this task even if the queue is full
+            return true;
+        }
     }
 
     public AliasFilter buildAliasFilter(ClusterState state, String index, Set<String> resolvedExpressions) {