fork client closing

Author: ywangd

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3PerProjectClientManager.java

@@ -23,23 +23,28 @@
 import org.elasticsearch.core.IOUtils;
 
 import java.io.Closeable;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
 public class S3PerProjectClientManager implements ClusterStateListener {
 
     private final Settings settings;
     private final Function<S3ClientSettings, AmazonS3> clientBuilder;
+    private final Executor executor;
     // A map of projectId to clients holder. Adding to and removing from the map happen only with the cluster state listener thread.
     private final Map<ProjectId, ClientsHolder> perProjectClientsCache;
 
-    public S3PerProjectClientManager(Settings settings, Function<S3ClientSettings, AmazonS3> clientBuilder) {
+    public S3PerProjectClientManager(Settings settings, Function<S3ClientSettings, AmazonS3> clientBuilder, Executor executor) {
         this.settings = settings;
         this.clientBuilder = clientBuilder;
+        this.executor = executor;
         this.perProjectClientsCache = new ConcurrentHashMap<>();
     }
 
@@ -70,22 +75,25 @@ public void clusterChanged(ClusterChangedEvent event) {
             }
         }
 
+        final List<ClientsHolder> clientsHoldersToClose = new ArrayList<>();
         // Updated projects
         for (var projectId : updatedPerProjectClients.keySet()) {
             final var old = perProjectClientsCache.put(projectId, updatedPerProjectClients.get(projectId));
             if (old != null) {
-                old.close();
+                clientsHoldersToClose.add(old);
             }
         }
-
         // removed projects
         for (var projectId : perProjectClientsCache.keySet()) {
             if (currentProjects.containsKey(projectId) == false) {
                 final var removed = perProjectClientsCache.remove(projectId);
                 assert removed != null;
-                removed.close();
+                clientsHoldersToClose.add(removed);
             }
         }
+        if (clientsHoldersToClose.isEmpty() == false) {
+            executor.execute(() -> IOUtils.closeWhileHandlingException(clientsHoldersToClose));
+        }
     }
 
     public AmazonS3Reference client(ProjectId projectId, RepositoryMetadata repositoryMetadata) {

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3RepositoryPlugin.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.plugins.RepositoryPlugin;
 import org.elasticsearch.repositories.RepositoriesMetrics;
 import org.elasticsearch.repositories.Repository;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 
@@ -91,7 +92,8 @@ public Collection<?> createComponents(PluginServices services) {
         if (services.projectResolver().supportsMultipleProjects()) {
             s3PerProjectClientManager = new S3PerProjectClientManager(
                 settings,
-                s3ClientSettings -> this.service.get().buildClient(s3ClientSettings)
+                s3ClientSettings -> this.service.get().buildClient(s3ClientSettings),
+                services.threadPool().executor(ThreadPool.Names.GENERIC)
             );
             services.clusterService().addListener(s3PerProjectClientManager);
         }