Add test assertions to ensure write bytes released

This is a follow-up to elastic#57573. This commit ensures that the bytes marked
in `WriteMemoryLimits` are released by any test using an internal test
cluster.

Author: Tim-Brooks

test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
 import org.elasticsearch.action.admin.indices.stats.CommonStatsFlags;
 import org.elasticsearch.action.admin.indices.stats.CommonStatsFlags.Flag;
+import org.elasticsearch.action.bulk.WriteMemoryLimits;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterName;
@@ -1154,10 +1155,34 @@ public void beforeIndexDeletion() throws Exception {
         // and not all docs have been purged after the test) and inherit from
         // ElasticsearchIntegrationTest must override beforeIndexDeletion() to avoid failures.
         assertNoPendingIndexOperations();
+        assertAllPendingWriteLimitsReleased();
         assertOpenTranslogReferences();
         assertNoSnapshottedIndexCommit();
     }
 
+    private void assertAllPendingWriteLimitsReleased() throws Exception {
+        assertBusy(() -> {
+            for (NodeAndClient nodeAndClient : nodes.values()) {
+                WriteMemoryLimits writeMemoryLimits = getInstance(WriteMemoryLimits.class, nodeAndClient.name);
+                final long coordinatingBytes = writeMemoryLimits.getCoordinatingBytes();
+                if (coordinatingBytes > 0) {
+                    throw new AssertionError("pending coordinating write bytes [" + coordinatingBytes + "] bytes on node ["
+                        + nodeAndClient.name + "].");
+                }
+                final long primaryBytes = writeMemoryLimits.getPrimaryBytes();
+                if (primaryBytes > 0) {
+                    throw new AssertionError("pending primary write bytes [" + coordinatingBytes + "] bytes on node ["
+                        + nodeAndClient.name + "].");
+                }
+                final long replicaBytes = writeMemoryLimits.getReplicaBytes();
+                if (replicaBytes > 0) {
+                    throw new AssertionError("pending replica write bytes [" + coordinatingBytes + "] bytes on node ["
+                        + nodeAndClient.name + "].");
+                }
+            }
+        }, 60, TimeUnit.SECONDS);
+    }
+
     private void assertNoPendingIndexOperations() throws Exception {
         assertBusy(() -> {
             for (NodeAndClient nodeAndClient : nodes.values()) {