Add test assertions to ensure write bytes released (#58970)

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

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/WriteMemoryLimitsIT.java

@@ -172,7 +172,7 @@ public class WriteMemoryLimitsIT extends ESIntegTestCase {
             final long secondBulkRequestSize = secondBulkRequest.ramBytesUsed();
             final long secondBulkShardRequestSize = request.ramBytesUsed();
 
-            assertEquals(bulkRequestSize + secondBulkRequestSize, replicaWriteLimits.getCoordinatingBytes());
+            assertBusy(() -> assertEquals(bulkRequestSize + secondBulkRequestSize, replicaWriteLimits.getCoordinatingBytes()));
             assertBusy(() -> assertThat(replicaWriteLimits.getReplicaBytes(),
                 greaterThan(bulkShardRequestSize + secondBulkShardRequestSize)));
 

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

@@ -37,6 +37,7 @@ import org.elasticsearch.action.admin.cluster.configuration.ClearVotingConfigExc
 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.client.transport.TransportClient;
@@ -1311,6 +1312,7 @@ public final class InternalTestCluster extends TestCluster {
         assertNoPendingIndexOperations();
         //check that shards that have same sync id also contain same number of documents
         assertSameSyncIdSameDocs();
+        assertAllPendingWriteLimitsReleased();
         assertOpenTranslogReferences();
         assertNoSnapshottedIndexCommit();
     }
@@ -1343,6 +1345,29 @@ public final class InternalTestCluster extends TestCluster {
         }
     }
 
+    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()) {