no client creation after manager closing

Author: ywangd

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

@@ -53,6 +53,7 @@ public class S3ClientsManager implements ClusterStateApplier {
     private final Settings nodeS3Settings;
     private final Function<S3ClientSettings, AmazonS3Reference> clientBuilder;
     private final Executor executor;
+    private final AtomicBoolean managerClosed = new AtomicBoolean(false);
     // A map of projectId to clients holder. Adding to and removing from the map happen only in the applier thread.
     private final Map<ProjectId, ClientsHolder<?>> clientsHolders;
 
@@ -211,7 +212,12 @@ void releaseCachedClients(ProjectId projectId) {
      * Shutdown the manager by closing all clients holders. This is called when the node is shutting down.
      */
     void close() {
-        IOUtils.closeWhileHandlingException(clientsHolders.values());
+        if (managerClosed.compareAndSet(false, true)) {
+            // Close all clients holders, they will close their cached clients.
+            // It's OK if a new clients holder is added concurrently or after this point because
+            // no new client will be created once the manager is closed, i.e. nothing to release.
+            IOUtils.closeWhileHandlingException(clientsHolders.values());
+        }
     }
 
     private boolean newOrUpdated(ProjectId projectId, Map<String, S3ClientSettings> currentClientSettings) {
@@ -292,6 +298,11 @@ final AmazonS3Reference client(RepositoryMetadata repositoryMetadata) {
                     // Not adding a new client once the manager is closed since there won't be anything to close it
                     throw new IllegalStateException("Project [" + projectId() + "] clients holder is closed");
                 }
+                if (managerClosed.get()) {
+                    // This clients holder must be added after the manager is closed. It must have no cached clients.
+                    assert clientsCache.isEmpty() : "expect empty cache, but got " + clientsCache;
+                    throw new IllegalStateException("s3 clients manager is closed");
+                }
                 // The close() method maybe called after we checked it, it is ok since we are already inside the synchronized block.
                 // The clearCache() will clear the newly added client.
                 final var newClientReference = clientBuilder.apply(settings);

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3ClientsManagerTests.java

@@ -42,10 +42,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -72,13 +70,11 @@ public class S3ClientsManagerTests extends ESTestCase {
     private ClusterService clusterService;
     private S3Service s3Service;
     private S3ClientsManager s3ClientsManager;
-    private final AtomicReference<CountDownLatch> clientRefsCloseLatchRef = new AtomicReference<>();
 
     @Override
     public void setUp() throws Exception {
         super.setUp();
         s3SecretsIdGenerators = ConcurrentCollections.newConcurrentMap();
-        clientRefsCloseLatchRef.set(null);
         clientNames = IntStream.range(0, between(2, 5)).mapToObj(i -> randomIdentifier() + "_" + i).toList();
 
         final Settings.Builder builder = Settings.builder();
@@ -296,6 +292,25 @@ public void testClusterAndProjectClients() {
         assertFalse(projectClient.hasReferences());
     }
 
+    public void testClientsHolderAfterManagerClosed() {
+        final ProjectId projectId = randomUniqueProjectId();
+        final String clientName = randomFrom(clientNames);
+
+        s3ClientsManager.close();
+        // New holder can be added after the manager is closed, but no actual client can be created
+        updateProjectInClusterState(projectId, newProjectClientsSecrets(projectId, clientName));
+        try (var clientsHolder = s3ClientsManager.getClientsHolders().get(projectId)) {
+            assertNotNull(clientsHolder);
+            assertFalse(clientsHolder.isClosed());
+
+            final IllegalStateException e = expectThrows(
+                IllegalStateException.class,
+                () -> s3ClientsManager.client(projectId, createRepositoryMetadata(clientName))
+            );
+            assertThat(e.getMessage(), containsString("s3 clients manager is closed"));
+        }
+    }
+
     public void testProjectClientsDisabled() {
         final var clusterService = spy(this.clusterService);
         final S3Service s3ServiceWithNoProjectSupport = new S3Service(