AWS: Do not create signer refresh executor when refresh is disabled (#7046)

Author: nastra

aws/src/main/java/org/apache/iceberg/aws/s3/signer/S3V4RestSignerClient.java

@@ -26,7 +26,6 @@
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -70,10 +69,13 @@ public abstract class S3V4RestSignerClient
   private static final Cache<Key, SignedComponent> SIGNED_COMPONENT_CACHE =
       Caffeine.newBuilder().expireAfterWrite(30, TimeUnit.SECONDS).maximumSize(100).build();
 
-  private static final ScheduledExecutorService TOKEN_REFRESH_EXECUTOR =
-      ThreadPools.newScheduledPool("s3-signer-token-refresh", 1);
   private static final String SCOPE = "sign";
-  private static final AtomicReference<RESTClient> HTTP_CLIENT_REF = new AtomicReference<>();
+
+  @SuppressWarnings("immutables:incompat")
+  private static volatile ScheduledExecutorService tokenRefreshExecutor;
+
+  @SuppressWarnings("immutables:incompat")
+  private static volatile RESTClient httpClient;
 
   public abstract Map<String, String> properties();
 
@@ -113,26 +115,44 @@ boolean keepTokenRefreshed() {
         OAuth2Properties.TOKEN_REFRESH_ENABLED_DEFAULT);
   }
 
+  private ScheduledExecutorService tokenRefreshExecutor() {
+    if (!keepTokenRefreshed()) {
+      return null;
+    }
+
+    if (null == tokenRefreshExecutor) {
+      synchronized (S3V4RestSignerClient.class) {
+        if (null == tokenRefreshExecutor) {
+          tokenRefreshExecutor = ThreadPools.newScheduledPool("s3-signer-token-refresh", 1);
+        }
+      }
+    }
+
+    return tokenRefreshExecutor;
+  }
+
   private RESTClient httpClient() {
-    if (null == HTTP_CLIENT_REF.get()) {
-      // TODO: should be closed
-      HTTP_CLIENT_REF.compareAndSet(
-          null,
-          HTTPClient.builder()
-              .uri(baseSignerUri())
-              .withObjectMapper(S3ObjectMapper.mapper())
-              .build());
+    if (null == httpClient) {
+      synchronized (S3V4RestSignerClient.class) {
+        if (null == httpClient) {
+          httpClient =
+              HTTPClient.builder()
+                  .uri(baseSignerUri())
+                  .withObjectMapper(S3ObjectMapper.mapper())
+                  .build();
+        }
+      }
     }
 
-    return HTTP_CLIENT_REF.get();
+    return httpClient;
   }
 
   private AuthSession authSession() {
     String token = token().get();
     if (null != token) {
       return AuthSession.fromAccessToken(
           httpClient(),
-          keepTokenRefreshed() ? TOKEN_REFRESH_EXECUTOR : null,
+          tokenRefreshExecutor(),
           token,
           expiresAtMillis(properties()),
           new AuthSession(ImmutableMap.of(), token, null, credential(), SCOPE));
@@ -144,11 +164,7 @@ private AuthSession authSession() {
       OAuthTokenResponse authResponse =
           OAuth2Util.fetchToken(httpClient(), session.headers(), credential(), SCOPE);
       return AuthSession.fromTokenResponse(
-          httpClient(),
-          keepTokenRefreshed() ? TOKEN_REFRESH_EXECUTOR : null,
-          authResponse,
-          startTimeMillis,
-          session);
+          httpClient(), tokenRefreshExecutor(), authResponse, startTimeMillis, session);
     }
 
     return AuthSession.empty();