[TEST] Add assertBusy when checking for pending operation counter after tests

Currently, pending operations can complete after tests with disruption scheme
completes. This commit waits for the pending operation counter to complete
after the tests are run

core/src/test/java/org/elasticsearch/discovery/DiscoveryWithServiceDisruptionsIT.java

@@ -175,7 +175,7 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
     }
 
     @Override
-    protected void beforeIndexDeletion() throws IOException {
+    protected void beforeIndexDeletion() throws Exception {
         if (disableBeforeIndexDeletion == false) {
             super.beforeIndexDeletion();
         }

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

@@ -577,7 +577,7 @@ public abstract class ESIntegTestCase extends ESTestCase {
         return Collections.emptySet();
     }
 
-    protected void beforeIndexDeletion() throws IOException {
+    protected void beforeIndexDeletion() throws Exception {
         cluster().beforeIndexDeletion();
     }
 

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

@@ -1020,7 +1020,7 @@ public final class InternalTestCluster extends TestCluster {
     }
 
     @Override
-    public void beforeIndexDeletion() throws IOException {
+    public void beforeIndexDeletion() throws Exception {
         // Check that the operations counter on index shard has reached 0.
         // The assumption here is that after a test there are no ongoing write operations.
         // test that have ongoing write operations after the test (for example because ttl is used
@@ -1055,7 +1055,8 @@ public final class InternalTestCluster extends TestCluster {
         }
     }
 
-    private void assertShardIndexCounter() throws IOException {
+    private void assertShardIndexCounter() throws Exception {
+        assertBusy(() -> {
             final Collection<NodeAndClient> nodesAndClients = nodes.values();
             for (NodeAndClient nodeAndClient : nodesAndClients) {
                 IndicesService indexServices = getInstance(IndicesService.class, nodeAndClient.name);
@@ -1070,7 +1071,9 @@ public final class InternalTestCluster extends TestCluster {
                                     .map(task -> task.taskInfo(localNode.getId(), true))
                                     .collect(Collectors.toList());
                             ListTasksResponse response = new ListTasksResponse(taskInfos, Collections.emptyList(), Collections.emptyList());
-                        XContentBuilder builder = XContentFactory.jsonBuilder()
+                            XContentBuilder builder = null;
+                            try {
+                                builder = XContentFactory.jsonBuilder()
                                         .prettyPrint()
                                         .startObject()
                                         .value(response)
@@ -1078,11 +1081,15 @@ public final class InternalTestCluster extends TestCluster {
                                 throw new AssertionError("expected index shard counter on shard " + indexShard.shardId() + " on node " +
                                         nodeAndClient.name + " to be 0 but was " + activeOperationsCount + ". Current replication tasks on node:\n" +
                                         builder.string());
+                            } catch (IOException e) {
+                                throw new RuntimeException("caught exception while building response [" + response + "]", e);
                             }
                         }
                     }
                 }
             }
+        });
+    }
 
     private void randomlyResetClients() throws IOException {
         // only reset the clients on nightly tests, it causes heavy load...

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

@@ -82,7 +82,7 @@ public abstract class TestCluster implements Closeable {
     /**
      * Assertions that should run before the cluster is wiped should be called in this method
      */
-    public void beforeIndexDeletion() throws IOException {
+    public void beforeIndexDeletion() throws Exception {
     }
 
     /**