Fix SearchableSnapshotDirectoryTests Leaking Listeners (#64150) (#64156)

We need to make sure we don't shut down the pool while ref count release
operations are enqueued but not yet executing.

Author: original-brownbear

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/index/store/SearchableSnapshotDirectoryTests.java

@@ -92,6 +92,7 @@
 import org.elasticsearch.test.IndexSettingsModule;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.threadpool.ThreadPoolStats;
 import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots;
 import org.elasticsearch.xpack.searchablesnapshots.cache.CacheService;
 import org.hamcrest.Matcher;
@@ -119,6 +120,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static java.util.Arrays.asList;
@@ -627,7 +629,7 @@ protected void assertSnapshotOrGenericThread() {
                 Releasables.close(releasables);
             }
         } finally {
-            terminate(threadPool);
+            terminateSafely(threadPool);
         }
     }
 
@@ -744,7 +746,7 @@ public void testClearCache() throws Exception {
                     }
                 }
             } finally {
-                terminate(threadPool);
+                terminateSafely(threadPool);
             }
         }
     }
@@ -940,6 +942,17 @@ private SearchableSnapshotRecoveryState createRecoveryState() {
         return recoveryState;
     }
 
+    // Wait for all operations on the threadpool to complete to make sure we don't leak any reference count releasing and then shut it down
+    private static void terminateSafely(ThreadPool threadPool) throws Exception {
+        assertBusy(() -> {
+            for (ThreadPoolStats.Stats stat : threadPool.stats()) {
+                assertEquals(stat.getActive(), 0);
+                assertEquals(stat.getQueue(), 0);
+            }
+        });
+        assertTrue(ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS));
+    }
+
     private static class FaultyReadsFileSystem extends FilterFileSystemProvider {
         FaultyReadsFileSystem(FileSystem inner) {
             super("faulty_fs://", inner);
@@ -956,5 +969,4 @@ public int read(ByteBuffer dst) throws IOException {
             };
         }
     }
-
 }