Attempt to fix ThreadLeakControl issues in S3BlobStoreRepositoryTests

Similar to opensearch-project#18201 but applied via the TestPlugin used in
S3BlobStoreRepositoryTests.

Closes opensearch-project#14299

Signed-off-by: David Causse <dcausse@wikimedia.org>

Author: nomoa

plugins/repository-s3/src/internalClusterTest/java/org/opensearch/repositories/s3/S3BlobStoreRepositoryTests.java

@@ -34,7 +34,7 @@
 import com.sun.net.httpserver.Headers;
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
-
+import fixture.s3.S3HttpHandler;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.ApplyTransactionIdStage;
 
 import org.opensearch.action.admin.indices.forcemerge.ForceMergeResponse;
@@ -74,14 +74,15 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
-import fixture.s3.S3HttpHandler;
-
-import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
-import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount;
 
 @SuppressForbidden(reason = "this test uses a HttpServer to emulate an S3 endpoint")
 // Need to set up a new cluster for each test because cluster settings use randomized authentication settings
@@ -234,9 +235,13 @@ public void testRequestStats() throws Exception {
      * S3RepositoryPlugin that allows to disable chunked encoding and to set a low threshold between single upload and multipart upload.
      */
     public static class TestS3RepositoryPlugin extends S3RepositoryPlugin {
-
         public TestS3RepositoryPlugin(final Settings settings, final Path configPath) {
-            super(settings, configPath);
+            super(
+                settings,
+                configPath,
+                new S3Service(configPath, Executors.newSingleThreadScheduledExecutor()),
+                new S3AsyncService(configPath, Executors.newSingleThreadScheduledExecutor())
+            );
         }
 
         @Override
@@ -246,6 +251,13 @@ public List<Setting<?>> getSettings() {
             return settings;
         }
 
+        @Override
+        public void close() throws IOException {
+            super.close();
+            Stream.of(service.getClientExecutorService(), s3AsyncService.getClientExecutorService())
+                .forEach(e -> assertTrue(ThreadPool.terminate(e, 5, TimeUnit.SECONDS)));
+        }
+
         @Override
         protected S3Repository createRepository(
             RepositoryMetadata metadata,

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3AsyncService.java

@@ -455,4 +455,9 @@ public void close() {
         releaseCachedClients();
 
     }
+
+    @Nullable
+    ScheduledExecutorService getClientExecutorService() {
+        return clientExecutorService;
+    }
 }

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3RepositoryPlugin.java

@@ -75,6 +75,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
@@ -106,6 +107,7 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin, Relo
     protected SizeBasedBlockingQ lowPrioritySizeBasedBlockingQ;
     protected TransferSemaphoresHolder transferSemaphoresHolder;
     protected GenericStatsMetricPublisher genericStatsMetricPublisher;
+    private ScheduledExecutorService scheduler;
 
     public S3RepositoryPlugin(final Settings settings, final Path configPath) {
         this(settings, configPath, new S3Service(configPath), new S3AsyncService(configPath));
@@ -234,6 +236,7 @@ public Collection<Object> createComponents(
 
         this.lowTransferQConsumerService = threadPool.executor(LOW_TRANSFER_QUEUE_CONSUMER);
         this.normalTransferQConsumerService = threadPool.executor(NORMAL_TRANSFER_QUEUE_CONSUMER);
+        this.scheduler = threadPool.scheduler();
 
         // High number of permit allocation because each op acquiring permit performs disk IO, computation and network IO.
         int availablePermits = Math.max(allocatedProcessors(clusterService.getSettings()) * 4, 10);

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3Service.java

@@ -527,4 +527,9 @@ public AwsCredentials resolveCredentials() {
     public void close() {
         releaseCachedClients();
     }
+
+    @Nullable
+    ScheduledExecutorService getClientExecutorService() {
+        return clientExecutorService;
+    }
 }