Core: Simplify AuthManager API

The current `AuthManager` interface exposes various methods for creating sessions: `initSession()`, `catalogSession()`, `contextualSession()`, `tableSession()`, etc.

Naming its methods after the intended usage of the returned session was, in hindsight, a questionable design choice: in some cases, the right method to call may not be so obvious, and if more session "flavors" are need in the future, more methods would need to be added to the interface.

This PR aims at fixing this by unifying all previous methods under one single method. The increased flexibility of the new method is due to its argument of type `AuthScope`, which can be freely implemented according to the caller's needs.

Author: adutra

aws/src/main/java/org/apache/iceberg/aws/RESTSigV4AuthManager.java

@@ -25,6 +25,8 @@
 import org.apache.iceberg.rest.RESTClient;
 import org.apache.iceberg.rest.RESTUtil;
 import org.apache.iceberg.rest.auth.AuthManager;
+import org.apache.iceberg.rest.auth.AuthScope;
+import org.apache.iceberg.rest.auth.AuthScopes;
 import org.apache.iceberg.rest.auth.AuthSession;
 import software.amazon.awssdk.auth.signer.Aws4Signer;
 
@@ -36,22 +38,58 @@
 @SuppressWarnings("unused") // loaded by reflection
 public class RESTSigV4AuthManager implements AuthManager {
 
-  private final Aws4Signer signer = Aws4Signer.create();
+  private final Aws4Signer signer;
   private final AuthManager delegate;
 
   private Map<String, String> catalogProperties = Map.of();
 
   public RESTSigV4AuthManager(String name, AuthManager delegate) {
     this.delegate = Preconditions.checkNotNull(delegate, "Invalid delegate: null");
+    signer = Aws4Signer.create();
   }
 
+  private RESTSigV4AuthManager(AuthManager delegate, Aws4Signer signer) {
+    this.delegate = Preconditions.checkNotNull(delegate, "Invalid delegate: null");
+    this.signer = signer;
+  }
+
+  @Override
+  public AuthManager withClient(RESTClient client) {
+    return new RESTSigV4AuthManager(delegate.withClient(client), signer);
+  }
+
+  @Override
+  public AuthSession authSession(AuthScope scope) {
+    AwsProperties awsProperties;
+    if (scope instanceof AuthScopes.Catalog) {
+      this.catalogProperties = scope.properties();
+      awsProperties = new AwsProperties(catalogProperties);
+    } else {
+      awsProperties = new AwsProperties(RESTUtil.merge(catalogProperties, scope.properties()));
+    }
+
+    return new RESTSigV4AuthSession(signer, delegate.authSession(scope), awsProperties);
+  }
+
+  /**
+   * @deprecated since 1.9.0, will be removed in 1.10.0; use {@link #authSession(AuthScope)}
+   *     instead.
+   */
   @Override
+  @Deprecated
+  @SuppressWarnings("deprecation")
   public RESTSigV4AuthSession initSession(RESTClient initClient, Map<String, String> properties) {
     return new RESTSigV4AuthSession(
         signer, delegate.initSession(initClient, properties), new AwsProperties(properties));
   }
 
+  /**
+   * @deprecated since 1.9.0, will be removed in 1.10.0; use {@link #authSession(AuthScope)}
+   *     instead.
+   */
   @Override
+  @Deprecated
+  @SuppressWarnings("deprecation")
   public RESTSigV4AuthSession catalogSession(
       RESTClient sharedClient, Map<String, String> properties) {
     this.catalogProperties = properties;
@@ -60,15 +98,27 @@ public RESTSigV4AuthSession catalogSession(
         signer, delegate.catalogSession(sharedClient, catalogProperties), awsProperties);
   }
 
+  /**
+   * @deprecated since 1.9.0, will be removed in 1.10.0; use {@link #authSession(AuthScope)}
+   *     instead.
+   */
   @Override
+  @Deprecated
+  @SuppressWarnings("deprecation")
   public AuthSession contextualSession(SessionCatalog.SessionContext context, AuthSession parent) {
     AwsProperties contextProperties =
         new AwsProperties(RESTUtil.merge(catalogProperties, context.properties()));
     return new RESTSigV4AuthSession(
         signer, delegate.contextualSession(context, parent), contextProperties);
   }
 
+  /**
+   * @deprecated since 1.9.0, will be removed in 1.10.0; use {@link #authSession(AuthScope)}
+   *     instead.
+   */
   @Override
+  @Deprecated
+  @SuppressWarnings("deprecation")
   public AuthSession tableSession(
       TableIdentifier table, Map<String, String> properties, AuthSession parent) {
     AwsProperties tableProperties =

aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java

@@ -32,6 +32,7 @@
 import org.apache.iceberg.rest.auth.AuthManager;
 import org.apache.iceberg.rest.auth.AuthManagers;
 import org.apache.iceberg.rest.auth.AuthSession;
+import org.apache.iceberg.rest.auth.ImmutableAuthScopes;
 import org.apache.iceberg.rest.credentials.Credential;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import software.amazon.awssdk.auth.credentials.AwsCredentials;
@@ -81,9 +82,11 @@ private RESTClient httpClient() {
     if (null == client) {
       synchronized (this) {
         if (null == client) {
-          authManager = AuthManagers.loadAuthManager("s3-credentials-refresh", properties);
           HTTPClient httpClient = HTTPClient.builder(properties).uri(properties.get(URI)).build();
-          authSession = authManager.catalogSession(httpClient, properties);
+          authManager =
+              AuthManagers.loadAuthManager("s3-credentials-refresh", properties)
+                  .withClient(httpClient);
+          authSession = authManager.authSession(ImmutableAuthScopes.Standalone.of(properties));
           client = httpClient.withAuthSession(authSession);
         }
       }

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

@@ -42,6 +42,7 @@
 import org.apache.iceberg.rest.auth.AuthManager;
 import org.apache.iceberg.rest.auth.AuthManagers;
 import org.apache.iceberg.rest.auth.AuthSession;
+import org.apache.iceberg.rest.auth.ImmutableAuthScopes;
 import org.apache.iceberg.rest.auth.OAuth2Properties;
 import org.apache.iceberg.rest.auth.OAuth2Util;
 import org.apache.iceberg.util.PropertyUtil;
@@ -156,7 +157,8 @@ AuthSession authSession() {
     if (null == authSession) {
       synchronized (S3V4RestSignerClient.class) {
         if (null == authSession) {
-          authManager = AuthManagers.loadAuthManager("s3-signer", properties());
+          authManager =
+              AuthManagers.loadAuthManager("s3-signer", properties()).withClient(httpClient());
           ImmutableMap.Builder<String, String> properties =
               ImmutableMap.<String, String>builder()
                   .putAll(properties())
@@ -173,7 +175,9 @@ AuthSession authSession() {
             properties.put(OAuth2Properties.CREDENTIAL, credential());
           }
 
-          authSession = authManager.tableSession(httpClient(), properties.buildKeepingLast());
+          authSession =
+              authManager.authSession(
+                  ImmutableAuthScopes.Standalone.of(properties.buildKeepingLast()));
         }
       }
     }

aws/src/test/java/org/apache/iceberg/aws/TestRESTSigV4AuthManager.java

@@ -20,23 +20,26 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.InstanceOfAssertFactories.type;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 
 import java.util.Map;
 import org.apache.iceberg.catalog.SessionCatalog;
 import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.rest.HTTPClient;
-import org.apache.iceberg.rest.RESTClient;
 import org.apache.iceberg.rest.auth.AuthManager;
 import org.apache.iceberg.rest.auth.AuthManagers;
 import org.apache.iceberg.rest.auth.AuthProperties;
 import org.apache.iceberg.rest.auth.AuthSession;
+import org.apache.iceberg.rest.auth.ImmutableAuthScopes;
 import org.apache.iceberg.rest.auth.NoopAuthManager;
 import org.apache.iceberg.rest.auth.OAuth2Manager;
 import org.apache.iceberg.rest.auth.OAuth2Util;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
 
 class TestRESTSigV4AuthManager {
 
@@ -106,64 +109,59 @@ void createInvalidCustomDelegate() {
   @Test
   void initSession() {
     AuthManager delegate = Mockito.mock(AuthManager.class);
-    when(delegate.initSession(any(), any())).thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
-    RESTClient client = Mockito.mock(RESTClient.class);
+    when(delegate.authSession(any())).thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
     AuthManager manager = new RESTSigV4AuthManager("test", delegate);
-    AuthSession authSession = manager.initSession(client, awsProperties);
-    assertThat(authSession)
-        .isInstanceOf(RESTSigV4AuthSession.class)
-        .extracting("delegate")
-        .isInstanceOf(OAuth2Util.AuthSession.class);
+    AuthSession authSession = manager.authSession(ImmutableAuthScopes.Initial.of(awsProperties));
+    assertAuthSession(authSession);
   }
 
   @Test
-  void catalogSession() {
+  void authSession() {
     AuthManager delegate = Mockito.mock(AuthManager.class);
-    when(delegate.catalogSession(any(), any()))
-        .thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
-    RESTClient client = Mockito.mock(RESTClient.class);
+    when(delegate.authSession(any())).thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
     AuthManager manager = new RESTSigV4AuthManager("test", delegate);
-    AuthSession authSession = manager.catalogSession(client, awsProperties);
-    assertThat(authSession)
-        .isInstanceOf(RESTSigV4AuthSession.class)
-        .extracting("delegate")
-        .isInstanceOf(OAuth2Util.AuthSession.class);
+    AuthSession authSession = manager.authSession(ImmutableAuthScopes.Catalog.of(awsProperties));
+    assertAuthSession(authSession);
   }
 
   @Test
   void contextualSession() {
     AuthManager delegate = Mockito.mock(AuthManager.class);
-    when(delegate.catalogSession(any(), any()))
-        .thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
-    when(delegate.contextualSession(any(), any()))
-        .thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
+    when(delegate.authSession(any())).thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
     AuthManager manager = new RESTSigV4AuthManager("test", delegate);
-    manager.catalogSession(Mockito.mock(HTTPClient.class), awsProperties);
+    AuthSession catalogSession = manager.authSession(ImmutableAuthScopes.Catalog.of(awsProperties));
     AuthSession authSession =
-        manager.contextualSession(
-            Mockito.mock(SessionCatalog.SessionContext.class), Mockito.mock(AuthSession.class));
-    assertThat(authSession)
-        .isInstanceOf(RESTSigV4AuthSession.class)
-        .extracting("delegate")
-        .isInstanceOf(OAuth2Util.AuthSession.class);
+        manager.authSession(
+            ImmutableAuthScopes.Contextual.of(
+                Mockito.mock(SessionCatalog.SessionContext.class), catalogSession));
+    assertAuthSession(authSession);
   }
 
   @Test
   void tableSession() {
     AuthManager delegate = Mockito.mock(AuthManager.class);
-    when(delegate.catalogSession(any(), any()))
-        .thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
-    when(delegate.tableSession(any(), any(), any()))
-        .thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
+    when(delegate.authSession(any())).thenReturn(Mockito.mock(OAuth2Util.AuthSession.class));
     AuthManager manager = new RESTSigV4AuthManager("test", delegate);
-    manager.catalogSession(Mockito.mock(HTTPClient.class), awsProperties);
+    AuthSession catalogSession = manager.authSession(ImmutableAuthScopes.Catalog.of(awsProperties));
     AuthSession authSession =
-        manager.tableSession(
-            Mockito.mock(TableIdentifier.class), Map.of(), Mockito.mock(AuthSession.class));
+        manager.authSession(
+            ImmutableAuthScopes.Table.of(
+                Mockito.mock(TableIdentifier.class), Map.of(), catalogSession));
+    assertAuthSession(authSession);
+  }
+
+  private static void assertAuthSession(AuthSession authSession) {
     assertThat(authSession)
         .isInstanceOf(RESTSigV4AuthSession.class)
         .extracting("delegate")
         .isInstanceOf(OAuth2Util.AuthSession.class);
+    assertThat(authSession).extracting("signer").isNotNull();
+    assertThat(authSession).extracting("signingRegion").isEqualTo(Region.of("us-west-2"));
+    assertThat(authSession)
+        .extracting("credentialsProvider")
+        .asInstanceOf(type(StaticCredentialsProvider.class))
+        .extracting(StaticCredentialsProvider::resolveCredentials)
+        .isEqualTo(AwsBasicCredentials.create("id", "secret"));
   }
 
   @Test

aws/src/test/java/org/apache/iceberg/aws/TestRESTSigV4Signer.java

@@ -32,6 +32,7 @@
 import org.apache.iceberg.rest.auth.AuthManagers;
 import org.apache.iceberg.rest.auth.AuthProperties;
 import org.apache.iceberg.rest.auth.AuthSession;
+import org.apache.iceberg.rest.auth.ImmutableAuthScopes;
 import org.apache.iceberg.rest.auth.OAuth2Util;
 import org.apache.iceberg.rest.responses.ConfigResponse;
 import org.apache.iceberg.rest.responses.OAuthTokenResponse;
@@ -78,8 +79,9 @@ public static void beforeClass() {
             .build()
             .withAuthSession(AuthSession.EMPTY);
 
-    authManager = AuthManagers.loadAuthManager("test", properties);
-    AuthSession authSession = authManager.catalogSession(httpClient, properties);
+    authManager = AuthManagers.loadAuthManager("test", properties).withClient(httpClient);
+    AuthSession authSession =
+        authManager.authSession(ImmutableAuthScopes.Standalone.of(properties));
 
     client = httpClient.withAuthSession(authSession);
   }