Catch exceptions from mapperService in StoreRecovery.recoverFromLocalShards (elastic#137077) (elastic#137114)

The inline listener there must not throw, or the recovery task will leak.

Author: bcully

docs/changelog/137077.yaml

@@ -0,0 +1,5 @@
+pr: 137077
+summary: Catch exceptions from `mapperService` in `StoreRecovery.recoverFromLocalShards`
+area: Recovery
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java

@@ -126,15 +126,16 @@ void recoverFromLocalShards(
                 mappingUpdateConsumer.accept(sourceMetadata.mapping(), mappingStep);
             }
             mappingStep.addListener(outerListener.delegateFailure((listener, ignored) -> {
-                indexShard.mapperService().merge(sourceMetadata, MapperService.MergeReason.MAPPING_RECOVERY);
-                // now that the mapping is merged we can validate the index sort configuration.
-                Sort indexSort = indexShard.getIndexSort();
-                final boolean hasNested = indexShard.mapperService().hasNested();
-                final boolean isSplit = sourceMetadata.getNumberOfShards() < indexShard.indexSettings().getNumberOfShards();
-
                 final var recoveryListener = recoveryListener(indexShard, listener);
-                logger.debug("starting recovery from local shards {}", shards);
+
                 try {
+                    indexShard.mapperService().merge(sourceMetadata, MapperService.MergeReason.MAPPING_RECOVERY);
+                    // now that the mapping is merged we can validate the index sort configuration.
+                    Sort indexSort = indexShard.getIndexSort();
+                    final boolean hasNested = indexShard.mapperService().hasNested();
+                    final boolean isSplit = sourceMetadata.getNumberOfShards() < indexShard.indexSettings().getNumberOfShards();
+
+                    logger.debug("starting recovery from local shards {}", shards);
                     final Directory directory = indexShard.store().directory(); // don't close this directory!!
                     final Directory[] sources = shards.stream().map(LocalShardSnapshot::getSnapshotDirectory).toArray(Directory[]::new);
                     final long maxSeqNo = shards.stream().mapToLong(LocalShardSnapshot::maxSeqNo).max().getAsLong();

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -203,6 +203,8 @@
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.oneOf;
 import static org.hamcrest.Matchers.sameInstance;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
 
 /**
  * Simple unit-test IndexShard related operations.
@@ -3374,6 +3376,14 @@ public void testRecoverFromLocalShard() throws IOException {
             });
             closeShards(differentIndex);
 
+            // check that an error from the mapper service is handled correctly
+            final PlainActionFuture<Boolean> badMapperFuture = new PlainActionFuture<>();
+            final IndexShard badMapper = spy(targetShard);
+            doThrow(IllegalArgumentException.class).when(badMapper).mapperService();
+            final BiConsumer<MappingMetadata, ActionListener<Void>> noopConsumer = (mapping, listener) -> listener.onResponse(null);
+            badMapper.recoverFromLocalShards(noopConsumer, List.of(sourceShard), badMapperFuture);
+            assertThrows(IndexShardRecoveryException.class, badMapperFuture::actionGet);
+
             final PlainActionFuture<Boolean> future = new PlainActionFuture<>();
             targetShard.recoverFromLocalShards(mappingConsumer, Arrays.asList(sourceShard), future);
             assertTrue(future.actionGet());