[7.13] Correctly log exceptions that are thrown during cache prewarming

#74606

Today if an exception is thrown during the prewarming of (parts of) 
the files it is simply swallowed. This is confusing because the shard 
recovery state is left in a FINALIZE state but nothing in the logs 
can help troubleshooting why the recovery has been interrupted.

This commit adds log traces to warn when an exception occurred 
during the prewarming of a part of a file.

Backport of #74419

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/SearchableSnapshotDirectory.java

@@ -506,8 +506,9 @@ private void prewarmCache(ActionListener<Void> listener) {
         }, listener::onFailure), snapshot().totalFileCount());
 
         for (BlobStoreIndexShardSnapshot.FileInfo file : snapshot().indexFiles()) {
-            if (file.metadata().hashEqualsContents() || isExcludedFromCache(file.physicalName())) {
-                if (file.metadata().hashEqualsContents()) {
+            boolean hashEqualsContents = file.metadata().hashEqualsContents();
+            if (hashEqualsContents || isExcludedFromCache(file.physicalName())) {
+                if (hashEqualsContents) {
                     recoveryState.getIndex().addFileDetail(file.physicalName(), file.length(), true);
                 } else {
                     recoveryState.ignoreFile(file.physicalName());
@@ -516,17 +517,24 @@ private void prewarmCache(ActionListener<Void> listener) {
                 continue;
             }
             recoveryState.getIndex().addFileDetail(file.physicalName(), file.length(), false);
+            boolean submitted = false;
             try {
                 final IndexInput input = openInput(file.physicalName(), CachedBlobContainerIndexInput.CACHE_WARMING_CONTEXT);
                 assert input instanceof CachedBlobContainerIndexInput : "expected cached index input but got " + input.getClass();
 
                 final int numberOfParts = file.numberOfParts();
                 final StepListener<Collection<Void>> fileCompletionListener = new StepListener<>();
-                fileCompletionListener.whenComplete(voids -> input.close(), e -> IOUtils.closeWhileHandlingException(input));
                 fileCompletionListener.addListener(completionListener.map(voids -> null));
+                fileCompletionListener.whenComplete(voids -> {
+                    logger.debug("{} file [{}] prewarmed", shardId, file.physicalName());
+                    input.close();
+                }, e -> {
+                    logger.warn(() -> new ParameterizedMessage("{} prewarming failed for file [{}]", shardId, file.physicalName()), e);
+                    IOUtils.closeWhileHandlingException(input);
+                });
 
                 final GroupedActionListener<Void> partsListener = new GroupedActionListener<>(fileCompletionListener, numberOfParts);
-
+                submitted = true;
                 for (int p = 0; p < numberOfParts; p++) {
                     final int part = p;
                     queue.add(Tuple.tuple(partsListener, () -> {
@@ -555,6 +563,9 @@ private void prewarmCache(ActionListener<Void> listener) {
                 }
             } catch (IOException e) {
                 logger.warn(() -> new ParameterizedMessage("{} unable to prewarm file [{}]", shardId, file.physicalName()), e);
+                if (submitted == false) {
+                    completionListener.onFailure(e);
+                }
             }
         }