Execute async cleanup tasks in CoordinatorTests (#91794)

Closing the `JoinValidationService` will enqueue a cache-cleaning task
which must be executed to release any pages held by the cache.

This commit also introduces a deterministic leak detector to replace the
one based on garbage collection, because in these tests we know exactly
the expected lifecycle of all allocated pages.

Closes #91599
Closes #91379
Closes #90576

test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java

@@ -53,7 +53,6 @@ import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.common.util.MockPageCacheRecycler;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
 import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
 import org.elasticsearch.common.util.set.Sets;
@@ -271,6 +270,7 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
         private final Map<Long, ClusterState> committedStatesByVersion = new HashMap<>();
         private final LinearizabilityChecker linearizabilityChecker = new LinearizabilityChecker();
         private final History history = new History();
+        private final CountingPageCacheRecycler countingPageCacheRecycler;
         private final Recycler<BytesRef> recycler;
         private final NodeHealthService nodeHealthService;
 
@@ -289,9 +289,8 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
 
         Cluster(int initialNodeCount, boolean allNodesMasterEligible, Settings nodeSettings, NodeHealthService nodeHealthService) {
             this.nodeHealthService = nodeHealthService;
-            this.recycler = usually()
-                ? BytesRefRecycler.NON_RECYCLING_INSTANCE
-                : new BytesRefRecycler(new MockPageCacheRecycler(Settings.EMPTY));
+            this.countingPageCacheRecycler = new CountingPageCacheRecycler();
+            this.recycler = new BytesRefRecycler(countingPageCacheRecycler);
             deterministicTaskQueue.setExecutionDelayVariabilityMillis(DEFAULT_DELAY_VARIABILITY);
 
             assertThat(initialNodeCount, greaterThan(0));
@@ -876,6 +875,12 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
             }
 
             clusterNodes.forEach(ClusterNode::close);
+
+            // Closing nodes may spawn some other background cleanup tasks that must also be run
+            runFor(DEFAULT_DELAY_VARIABILITY, "accumulate close-time tasks");
+            deterministicTaskQueue.runAllRunnableTasks();
+
+            countingPageCacheRecycler.assertAllPagesReleased();
         }
 
         protected List<NamedWriteableRegistry.Entry> extraNamedWriteables() {

test/framework/src/main/java/org/elasticsearch/cluster/coordination/CountingPageCacheRecycler.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.cluster.coordination;
+
+import org.elasticsearch.common.recycler.Recycler;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.PageCacheRecycler;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class CountingPageCacheRecycler extends PageCacheRecycler {
+
+    private int openPages = 0;
+
+    public CountingPageCacheRecycler() {
+        super(Settings.EMPTY);
+    }
+
+    @Override
+    public Recycler.V<byte[]> bytePage(boolean clear) {
+        final var page = super.bytePage(clear);
+        openPages += 1;
+        return new Recycler.V<>() {
+            boolean closed = false;
+
+            @Override
+            public byte[] v() {
+                return page.v();
+            }
+
+            @Override
+            public boolean isRecycled() {
+                return page.isRecycled();
+            }
+
+            @Override
+            public void close() {
+                assertFalse(closed);
+                closed = true;
+                openPages -= 1;
+                page.close();
+            }
+        };
+    }
+
+    @Override
+    public Recycler.V<Object[]> objectPage() {
+        throw new AssertionError("unexpected call to objectPage()");
+    }
+
+    public void assertAllPagesReleased() {
+        assertEquals(0, openPages);
+    }
+}