Write multiple cache gaps for warming service in one go (elastic#112071)

* Write multiple cache gaps for warming service in one go

Author: volodk85

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java

@@ -967,12 +967,48 @@ void populate(
                         listener.onResponse(false);
                         return;
                     }
-                    try (var gapsListener = new RefCountingListener(listener.map(unused -> true))) {
-                        assert writer.sharedInputStreamFactory(gaps) == null;
-                        for (SparseFileTracker.Gap gap : gaps) {
-                            executor.execute(
-                                fillGapRunnable(gap, writer, null, ActionListener.releaseAfter(gapsListener.acquire(), refs.acquire()))
-                            );
+                    final SourceInputStreamFactory streamFactory = writer.sharedInputStreamFactory(gaps);
+                    logger.trace(
+                        () -> Strings.format(
+                            "fill gaps %s %s shared input stream factory",
+                            gaps,
+                            streamFactory == null ? "without" : "with"
+                        )
+                    );
+                    if (streamFactory == null) {
+                        try (var parallelGapsListener = new RefCountingListener(listener.map(unused -> true))) {
+                            for (SparseFileTracker.Gap gap : gaps) {
+                                executor.execute(
+                                    fillGapRunnable(
+                                        gap,
+                                        writer,
+                                        null,
+                                        ActionListener.releaseAfter(parallelGapsListener.acquire(), refs.acquire())
+                                    )
+                                );
+                            }
+                        }
+                    } else {
+                        try (
+                            var sequentialGapsListener = new RefCountingListener(
+                                ActionListener.runBefore(listener.map(unused -> true), streamFactory::close)
+                            )
+                        ) {
+                            final List<Runnable> gapFillingTasks = gaps.stream()
+                                .map(
+                                    gap -> fillGapRunnable(
+                                        gap,
+                                        writer,
+                                        streamFactory,
+                                        ActionListener.releaseAfter(sequentialGapsListener.acquire(), refs.acquire())
+                                    )
+                                )
+                                .toList();
+                            executor.execute(() -> {
+                                // Fill the gaps in order. If a gap fails to fill for whatever reason, the task for filling the next
+                                // gap will still be executed.
+                                gapFillingTasks.forEach(Runnable::run);
+                            });
                         }
                     }
                 }
@@ -1019,8 +1055,7 @@ void populateAndRead(
                             () -> Strings.format(
                                 "fill gaps %s %s shared input stream factory",
                                 gaps,
-                                (streamFactory == null ? "without" : "with"),
-                                (streamFactory == null ? "" : " " + streamFactory)
+                                streamFactory == null ? "without" : "with"
                             )
                         );
                         if (streamFactory == null) {

x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBlobCacheServiceTests.java

@@ -1422,7 +1422,7 @@ protected int computeCacheFileRegionSize(long fileLength, int region) {
         }
     }
 
-    public void testSharedSourceInputStreamFactory() throws Exception {
+    public void testUsageSharedSourceInputStreamFactoryInCachePopulation() throws Exception {
         final long regionSizeInBytes = size(100);
         final Settings settings = Settings.builder()
             .put(NODE_NAME_SETTING.getKey(), "node")
@@ -1519,16 +1519,22 @@ public void fillCacheRange(
             };
 
             final var range = ByteRange.of(0, regionSizeInBytes);
-            final PlainActionFuture<Integer> future = new PlainActionFuture<>();
-            region.populateAndRead(
-                range,
-                range,
-                (channel, channelPos, relativePos, length) -> length,
-                rangeMissingHandler,
-                threadPool.generic(),
-                future
-            );
-            safeGet(future);
+            if (randomBoolean()) {
+                final PlainActionFuture<Integer> future = new PlainActionFuture<>();
+                region.populateAndRead(
+                    range,
+                    range,
+                    (channel, channelPos, relativePos, length) -> length,
+                    rangeMissingHandler,
+                    threadPool.generic(),
+                    future
+                );
+                assertThat(safeGet(future).longValue(), equalTo(regionSizeInBytes));
+            } else {
+                final PlainActionFuture<Boolean> future = new PlainActionFuture<>();
+                region.populate(range, rangeMissingHandler, threadPool.generic(), future);
+                assertThat(safeGet(future), equalTo(true));
+            }
             assertThat(invocationCounter.get(), equalTo(numberGaps));
             assertThat(region.tracker.checkAvailable(regionSizeInBytes), is(true));
             assertBusy(() -> assertThat(factoryClosed.get(), is(true)));