Azure: Support vended credentials refresh in ADLSFileIO.

[ChaladiMohanVamsi]

Proposed Change

Add support to refresh and consume vended storage credentials for ADLSFileIO.

New Azure properties

• client.refresh-credentials-enabled property to control credential refresh and defaults to true.
• client.refresh-credentials-endpoint defines endpoint to fetch the refresh credentials with below spec.
  • OpenAPI: Add endpoint to retrieve valid credentials for a given table #11281

Requirements of credential config

• Required to have Sas token expiration time as part of credential config along with Sas token.
• Expected prefix for the expiration time is adls.sas-token-expire-at-ms. similar to existing sas token prefix adls.sas-token.
• Expected to have only one ADLS credential per storage-account prefix.

Similar PRs for other FileIOs

• AWS: Refresh vended credentials #11389
• GCS: Refresh vended credentials #11282

[ChaladiMohanVamsi]

cc// @nastra @jackye1995 @amogh-jahagirdar @munendrasn Can you please help with the review.

[nastra]

can you elaborate why this is needed here?

[ChaladiMohanVamsi]

Tried to implement the similar behaviour present in RESTSessionCatalog, where catalog can be configured to pass explicit headers to server by setting the configuration with header. prefix.

[nastra]

does this actually need to be a supplier given that it's being immediately used in L40?

[ChaladiMohanVamsi]

Yes, the supplier is being used during initialization and also during each scheduled refresh to fetch the new credentials. Supplier holds the logic to fetch new credentials from the API endpoint, since we are going to use it multiple times I modelled it as supplier instead of single method call. Please suggest if there is a cleaner way to achieve the same.

[nastra]

I'm not sure what this comment is trying to say. Also what happens if you remove the sleep time above?

[ChaladiMohanVamsi]

• Since the credential refresh implementation is based on scheduling a new refresh using the server provided expiration time,
• this test is covering the scenario where server responds with already expired token (expires-at-ms < current-time).
• The scheduling delay for consequent refresh is minimum delay of 10ms as per the logic in refreshDelayMillis()
• If we remove the sleep time, the credentials will be fetched from API only once. Depending on the test case execution time (> 10ms) there can be second credential fetch. To keep the test case behaviour consistent I used thread sleep.

[nastra]

@ChaladiMohanVamsi thanks for working on this. Do you have a way of actually testing this PR with an ADLS environment and see whether the refreshes work?

[ChaladiMohanVamsi]

@nastra @amogh-jahagirdar Thanks for the review and suggestions. I have addressed the review comments.

I was able to test the credentials refresh logic in ADLS environment, since I don't have the new credentials API endpoint spec implementation in my env I used loadTable endpoint for testing by tweaking few parts of this PR.

[danielcweeks]

This is going to change with the introduction of the new Auth manager, so I would wait until that goes in then switch over to using that approach. See how this is done for S3 here

[ChaladiMohanVamsi]

Incorporated the Auth manager releated changes referring to the changes for S3 FileIO and GCS FileIO.

cc// @adutra can you please review the changes.

[ajantha-bhat]

@ChaladiMohanVamsi: Are you still working on this PR? @nastra has added 1.9.0 milestone for this. I was thinking a 1.9.0 release cut tomorrow evening if everything is merged.

[ChaladiMohanVamsi]

@ChaladiMohanVamsi: Are you still working on this PR? @nastra has added 1.9.0 milestone for this. I was thinking a 1.9.0 release cut tomorrow evening if everything is merged.

Yes @ajantha-bhat, I will try to address the comments by today.

[ChaladiMohanVamsi]

@nastra @danielcweeks can you please review on the latest changes handling review comments.

[nastra]

I think we can remove this, since we're already copying over allProperties into credentialProviderProperties

[ChaladiMohanVamsi]

removed redundant operation of setting token.

[nastra]

nit: newline after }

[ChaladiMohanVamsi]

applied

[nastra]

can you please apply the below diff to this class?

--- a/azure/src/main/java/org/apache/iceberg/azure/adlsv2/VendedAdlsCredentialProvider.java
+++ b/azure/src/main/java/org/apache/iceberg/azure/adlsv2/VendedAdlsCredentialProvider.java
@@ -28,6 +28,7 @@ import java.time.ZoneOffset;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.azure.AzureProperties;
 import org.apache.iceberg.io.CloseableGroup;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -48,6 +49,8 @@ public class VendedAdlsCredentialProvider implements Serializable, AutoCloseable
   public static final String URI = "credentials.uri";

   private final SerializableMap<String, String> properties;
+  private final String credentialsEndpoint;
+  private final String catalogEndpoint;
   private transient volatile Map<String, SimpleTokenCache> sasCredentialByAccount;
   private transient volatile HTTPClient client;
   private transient AuthManager authManager;
@@ -55,8 +58,12 @@ public class VendedAdlsCredentialProvider implements Serializable, AutoCloseable

   public VendedAdlsCredentialProvider(Map<String, String> properties) {
     Preconditions.checkArgument(null != properties, "Invalid properties: null");
-    Preconditions.checkArgument(null != properties.get(URI), "Invalid URI: null");
+    Preconditions.checkArgument(null != properties.get(URI), "Invalid credentials endpoint: null");
+    Preconditions.checkArgument(
+        null != properties.get(CatalogProperties.URI), "Invalid catalog endpoint: null");
     this.properties = SerializableMap.copyOf(properties);
+    this.credentialsEndpoint = properties.get(URI);
+    this.catalogEndpoint = properties.get(CatalogProperties.URI);
   }

   String credentialForAccount(String storageAccount) {
@@ -117,7 +124,7 @@ public class VendedAdlsCredentialProvider implements Serializable, AutoCloseable
       synchronized (this) {
         if (null == client) {
           authManager = AuthManagers.loadAuthManager("adls-credentials-refresh", properties);
-          HTTPClient httpClient = HTTPClient.builder(properties).uri(properties.get(URI)).build();
+          HTTPClient httpClient = HTTPClient.builder(properties).uri(catalogEndpoint).build();
           authSession = authManager.catalogSession(httpClient, properties);
           client = httpClient.withAuthSession(authSession);
         }
@@ -130,7 +137,7 @@ public class VendedAdlsCredentialProvider implements Serializable, AutoCloseable
   private LoadCredentialsResponse fetchCredentials() {
     return httpClient()
         .get(
-            properties.get(URI),
+            credentialsEndpoint,
             null,

This is fixing an issue that we're currently also fixing for S3 / GCP in #12612 / #12638. You will also have to update the tests in a similar way as in those PRs

[ChaladiMohanVamsi]

Applied the patch and associated test cases.

[nastra]

I believe we should be able to make this non-public

[ChaladiMohanVamsi]

updated it to non-public.

[nastra]

overall LGTM once the remaining comments have been applied

[danielcweeks]

+1 pending checks. Thanks @ChaladiMohanVamsi !

[nastra]

thanks @ChaladiMohanVamsi

[tedyu]

Should this be enclosed in finally block so that it is always run ?