Add a latch to sync when all max upload tasks have been submitted (#101203)

Before we check the amount of active tasks on the prewarming executor,
we need to verify that all the tasks have been actually submitted.

Otherwise, we have a race in and amount of active tasks can be lower
then the amount of submitted tasks.

Fixes #99124

---------

Co-authored-by: David Turner <david.turner@elastic.co>

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/cache/full/SearchableSnapshotsPrewarmingIntegTests.java

@@ -251,11 +251,13 @@ public class SearchableSnapshotsPrewarmingIntegTests extends ESSingleNodeTestCas
         final CountDownLatch startPrewarmingLatch = new CountDownLatch(1);
         final var threadPool = getInstanceFromNode(ThreadPool.class);
         final int maxUploadTasks = threadPool.info(CACHE_PREWARMING_THREAD_POOL_NAME).getMax();
+        final CountDownLatch maxUploadTasksCreated = new CountDownLatch(maxUploadTasks);
         for (int i = 0; i < maxUploadTasks; i++) {
             threadPool.executor(CACHE_PREWARMING_THREAD_POOL_NAME).execute(new AbstractRunnable() {
 
                 @Override
                 protected void doRun() throws Exception {
+                    maxUploadTasksCreated.countDown();
                     startPrewarmingLatch.await();
                 }
 
@@ -265,7 +267,7 @@ public class SearchableSnapshotsPrewarmingIntegTests extends ESSingleNodeTestCas
                 }
             });
         }
-
+        safeAwait(maxUploadTasksCreated);
         var prewarmingExecutor = threadPool.executor(CACHE_PREWARMING_THREAD_POOL_NAME);
         assertThat(prewarmingExecutor, instanceOf(ThreadPoolExecutor.class));
         assertThat(((ThreadPoolExecutor) prewarmingExecutor).getActiveCount(), equalTo(maxUploadTasks));