Correctly log exceptions that are thrown during cache prewarming (#74419)

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.

Author: tlrx

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);
+ }
  }
  }