Safely Close Repositories on Node Shutdown (#48020)

server/src/main/java/org/elasticsearch/node/Node.java

@@ -668,6 +668,7 @@ public Node start() throws NodeValidationException {
         injector.getInstance(IndicesClusterStateService.class).start();
         injector.getInstance(SnapshotsService.class).start();
         injector.getInstance(SnapshotShardsService.class).start();
+        injector.getInstance(RepositoriesService.class).start();
         injector.getInstance(SearchService.class).start();
         nodeService.getMonitorService().start();
 
@@ -780,6 +781,7 @@ private Node stop() {
 
         injector.getInstance(SnapshotsService.class).stop();
         injector.getInstance(SnapshotShardsService.class).stop();
+        injector.getInstance(RepositoriesService.class).stop();
         // stop any changes happening as a result of cluster state changes
         injector.getInstance(IndicesClusterStateService.class).stop();
         // close discovery early to not react to pings anymore.

server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java

@@ -37,15 +37,19 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.snapshots.RestoreService;
 import org.elasticsearch.snapshots.SnapshotsService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -54,7 +58,7 @@
 /**
  * Service responsible for maintaining and providing access to snapshot repositories on nodes.
  */
-public class RepositoriesService implements ClusterStateApplier {
+public class RepositoriesService extends AbstractLifecycleComponent implements ClusterStateApplier {
 
     private static final Logger logger = LogManager.getLogger(RepositoriesService.class);
 
@@ -95,6 +99,8 @@ public RepositoriesService(Settings settings, ClusterService clusterService, Tra
      * @param listener register repository listener
      */
     public void registerRepository(final PutRepositoryRequest request, final ActionListener<ClusterStateUpdateResponse> listener) {
+        assert lifecycle.started() : "Trying to register new repository but service is in state [" + lifecycle.state() + "]";
+
         final RepositoryMetaData newRepositoryMetaData = new RepositoryMetaData(request.name(), request.type(), request.settings());
         validate(request.name());
 
@@ -435,4 +441,23 @@ private void ensureRepositoryNotInUse(ClusterState clusterState, String reposito
             throw new IllegalStateException("trying to modify or unregister repository that is currently used ");
         }
     }
+
+    @Override
+    protected void doStart() {
+
+    }
+
+    @Override
+    protected void doStop() {
+
+    }
+
+    @Override
+    protected void doClose() throws IOException {
+        clusterService.removeApplier(this);
+        final Collection<Repository> repos = new ArrayList<>();
+        repos.addAll(internalRepositories.values());
+        repos.addAll(repositories.values());
+        IOUtils.close(repos);
+    }
 }

server/src/test/java/org/elasticsearch/repositories/RepositoriesServiceTests.java

@@ -67,6 +67,7 @@ public void setUp() throws Exception {
             Collections.emptySet());
         repositoriesService = new RepositoriesService(Settings.EMPTY, mock(ClusterService.class),
             transportService, Collections.emptyMap(), Collections.singletonMap(TestRepository.TYPE, TestRepository::new), threadPool);
+        repositoriesService.start();
     }
 
     public void testRegisterInternalRepository() {

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -1186,6 +1186,7 @@ public void start(ClusterState initialState) {
                 transportService.acceptIncomingRequests();
                 snapshotsService.start();
                 snapshotShardsService.start();
+                repositoriesService.start();
                 final CoordinationState.PersistedState persistedState =
                     new InMemoryPersistedState(initialState.term(), stateForNode(initialState, node));
                 coordinator = new Coordinator(node.getName(), clusterService.getSettings(),

x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SLMSnapshotBlockingIntegTests.java

@@ -127,7 +127,6 @@ protected Settings transportClientSettings() {
         return settings.build();
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/47689")
     public void testSnapshotInProgress() throws Exception {
         final String indexName = "test";
         final String policyName = "test-policy";