Improve watcher cleanup logic (#67554)

by waiting for all tasks to complete before deleting watcher history data stream.

Closes #66495

Author: martijnvg

x-pack/plugin/watcher/src/internalClusterTest/java/org/elasticsearch/xpack/watcher/test/AbstractWatcherIntegrationTestCase.java

@@ -202,6 +202,11 @@ public void _cleanup() throws Exception {
         // Clear all internal watcher state for the next test method:
         logger.info("[#{}]: clearing watcher state", getTestName());
         stopWatcher();
+        // Wait for all pending tasks to complete, this to avoid any potential incoming writes
+        // to watcher history data stream to recreate the data stream after it has been created.
+        // Otherwise ESIntegTestCase test cluster's wipe cluster logic that deletes all indices may fail,
+        // because it attempts to remove the write index of an existing data stream.
+        waitNoPendingTasksOnAll();
         String[] dataStreamsToDelete = {HistoryStoreField.DATA_STREAM};
         client().execute(DeleteDataStreamAction.INSTANCE, new DeleteDataStreamAction.Request(dataStreamsToDelete));
         GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(dataStreamsToDelete);