Fix ThreadPoolMergeExecutorServiceTests testIORateIsAdjustedForAllRunningMergeTasks (elastic#130545) (elastic#130560)

The test submits merge tasks that support IO throttling, and asserts
that all the currently running merge tasks are indeed IO throttled after
the new one was submitted.

The test erroneously tried to assert a property on the set of currently
running merge tasks, which is very difficult to do since all merge tasks
are possibly backlogged and re-enqueued asynchronously multiple times
before they are run or aborted (so looking at the threadpool merge task
queue there's no telling which merge task will execute first).

Fixes elastic#129531

Author: albertzaharovits

muted-tests.yml

@@ -491,9 +491,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
   method: test {knn-function.KnnSearchWithKOption SYNC}
   issue: https://github.com/elastic/elasticsearch/issues/129512
-- class: org.elasticsearch.index.engine.ThreadPoolMergeExecutorServiceTests
-  method: testIORateIsAdjustedForAllRunningMergeTasks
-  issue: https://github.com/elastic/elasticsearch/issues/129531
 - class: org.elasticsearch.upgrades.IndexingIT
   method: testIndexing
   issue: https://github.com/elastic/elasticsearch/issues/129533

server/src/test/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorServiceTests.java

@@ -51,7 +51,6 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
@@ -329,7 +328,6 @@ public void testIORateIsAdjustedForAllRunningMergeTasks() throws Exception {
             ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) testThreadPool.executor(ThreadPool.Names.MERGE);
             Semaphore runMergeSemaphore = new Semaphore(0);
             Set<MergeTask> currentlyRunningMergeTasksSet = ConcurrentCollections.newConcurrentSet();
-            Set<MergeTask> currentlyRunningOrAbortingMergeTasksSet = ConcurrentCollections.newConcurrentSet();
             while (mergesStillToComplete > 0) {
                 if (mergesStillToSubmit > 0 && (currentlyRunningMergeTasksSet.isEmpty() || randomBoolean())) {
                     MergeTask mergeTask = mock(MergeTask.class);
@@ -347,27 +345,17 @@ public void testIORateIsAdjustedForAllRunningMergeTasks() throws Exception {
                     }).when(mergeTask).schedule();
                     doAnswer(mock -> {
                         currentlyRunningMergeTasksSet.add(mergeTask);
-                        currentlyRunningOrAbortingMergeTasksSet.add(mergeTask);
                         // wait to be signalled before completing
                         runMergeSemaphore.acquire();
-                        currentlyRunningOrAbortingMergeTasksSet.remove(mergeTask);
                         currentlyRunningMergeTasksSet.remove(mergeTask);
                         return null;
                     }).when(mergeTask).run();
                     doAnswer(mock -> {
-                        currentlyRunningOrAbortingMergeTasksSet.add(mergeTask);
                         // wait to be signalled before completing
                         runMergeSemaphore.acquire();
-                        currentlyRunningOrAbortingMergeTasksSet.remove(mergeTask);
                         return null;
                     }).when(mergeTask).abort();
-                    assertThat(runMergeSemaphore.availablePermits(), is(0));
-                    boolean isAnyExecutorAvailable = currentlyRunningOrAbortingMergeTasksSet.size() < mergeExecutorThreadCount;
-                    boolean mergeTaskSubmitted = threadPoolMergeExecutorService.submitMergeTask(mergeTask);
-                    assertTrue(mergeTaskSubmitted);
-                    if (isAnyExecutorAvailable) {
-                        assertBusy(() -> assertThat(currentlyRunningOrAbortingMergeTasksSet, hasItem(mergeTask)));
-                    }
+                    assertTrue(threadPoolMergeExecutorService.submitMergeTask(mergeTask));
                     long latestIORate = threadPoolMergeExecutorService.getTargetIORateBytesPerSec();
                     // all currently running merge tasks must be IO throttled to the latest IO Rate
                     assertBusy(() -> {