HADOOP-18542. Keep MSI tenant ID and client ID optional (#4262)

Contributed by Carl Levasseur

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java

@@ -1084,9 +1084,9 @@ public AccessTokenProvider getTokenProvider() throws TokenAccessProviderExceptio
               FS_AZURE_ACCOUNT_OAUTH_MSI_ENDPOINT,
               AuthConfigurations.DEFAULT_FS_AZURE_ACCOUNT_OAUTH_MSI_ENDPOINT);
           String tenantGuid =
-              getMandatoryPasswordString(FS_AZURE_ACCOUNT_OAUTH_MSI_TENANT);
+              getPasswordString(FS_AZURE_ACCOUNT_OAUTH_MSI_TENANT);
           String clientId =
-              getMandatoryPasswordString(FS_AZURE_ACCOUNT_OAUTH_CLIENT_ID);
+              getPasswordString(FS_AZURE_ACCOUNT_OAUTH_CLIENT_ID);
           String authority = getTrimmedPasswordString(
               FS_AZURE_ACCOUNT_OAUTH_MSI_AUTHORITY,
               AuthConfigurations.DEFAULT_FS_AZURE_ACCOUNT_OAUTH_MSI_AUTHORITY);

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/TestAccountConfiguration.java

@@ -27,6 +27,7 @@
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.ConfigurationPropertyNotFoundException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidConfigurationValueException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.TokenAccessProviderException;
+import org.apache.hadoop.fs.azurebfs.oauth2.AccessTokenProvider;
 import org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider;
 import org.apache.hadoop.fs.azurebfs.oauth2.CustomTokenProviderAdapter;
 import org.apache.hadoop.fs.azurebfs.oauth2.MsiTokenProvider;
@@ -66,6 +67,7 @@
  */
 public class TestAccountConfiguration {
   private static final String TEST_OAUTH_PROVIDER_CLASS_CONFIG = "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider";
+  private static final String TEST_OAUTH_MSI_TOKEN_PROVIDER_CLASS_CONFIG = "org.apache.hadoop.fs.azurebfs.oauth2.MsiTokenProvider";
   private static final String TEST_CUSTOM_PROVIDER_CLASS_CONFIG = "org.apache.hadoop.fs.azurebfs.oauth2.RetryTestTokenProvider";
   private static final String TEST_SAS_PROVIDER_CLASS_CONFIG_1 = "org.apache.hadoop.fs.azurebfs.extensions.MockErrorSASTokenProvider";
   private static final String TEST_SAS_PROVIDER_CLASS_CONFIG_2 = "org.apache.hadoop.fs.azurebfs.extensions.MockSASTokenProvider";
@@ -90,11 +92,6 @@ public class TestAccountConfiguration {
           FS_AZURE_ACCOUNT_OAUTH_USER_NAME,
           FS_AZURE_ACCOUNT_OAUTH_USER_PASSWORD));
 
-  private static final List<String> MSI_TOKEN_OAUTH_CONFIG_KEYS =
-      Collections.unmodifiableList(Arrays.asList(
-          FS_AZURE_ACCOUNT_OAUTH_MSI_TENANT,
-          FS_AZURE_ACCOUNT_OAUTH_CLIENT_ID));
-
   private static final List<String> REFRESH_TOKEN_OAUTH_CONFIG_KEYS =
       Collections.unmodifiableList(Arrays.asList(
           FS_AZURE_ACCOUNT_OAUTH_REFRESH_TOKEN,
@@ -410,10 +407,8 @@ public void testAccessTokenProviderPrecedence()
   public void testOAuthConfigPropNotFound() throws Throwable {
     testConfigPropNotFound(CLIENT_CREDENTIAL_OAUTH_CONFIG_KEYS, ClientCredsTokenProvider.class.getName());
     testConfigPropNotFound(USER_PASSWORD_OAUTH_CONFIG_KEYS, UserPasswordTokenProvider.class.getName());
-    testConfigPropNotFound(MSI_TOKEN_OAUTH_CONFIG_KEYS, MsiTokenProvider.class.getName());
     testConfigPropNotFound(REFRESH_TOKEN_OAUTH_CONFIG_KEYS, RefreshTokenBasedTokenProvider.class.getName());
     testConfigPropNotFound(WORKLOAD_IDENTITY_OAUTH_CONFIG_KEYS, WorkloadIdentityTokenProvider.class.getName());
-
   }
 
   private void testConfigPropNotFound(List<String> configKeys,
@@ -444,6 +439,30 @@ private static void testMissingConfigKey(final AbfsConfiguration abfsConf,
                 () -> abfsConf.getTokenProvider().getClass().getTypeName())));
   }
 
+  @Test
+  public void testClientAndTenantIdOptionalWhenUsingMsiTokenProvider() throws Throwable {
+      final String accountName = "account";
+      final Configuration conf = new Configuration();
+      final AbfsConfiguration abfsConf = new AbfsConfiguration(conf, accountName);
+
+      final String accountNameSuffix = "." + abfsConf.getAccountName();
+      String authKey = FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME + accountNameSuffix;
+      String providerClassKey = "";
+      String providerClassValue = "";
+
+      providerClassKey = FS_AZURE_ACCOUNT_TOKEN_PROVIDER_TYPE_PROPERTY_NAME + accountNameSuffix;
+      providerClassValue = TEST_OAUTH_MSI_TOKEN_PROVIDER_CLASS_CONFIG;
+
+      abfsConf.set(authKey, AuthType.OAuth.toString());
+      abfsConf.set(providerClassKey, providerClassValue);
+
+      AccessTokenProvider tokenProviderTypeName = abfsConf.getTokenProvider();
+      // Test that we managed to instantiate an MsiTokenProvider without having to define the tenant and client ID.
+      // Those 2 fields are optional as they can automatically be determined by the Azure Metadata service when
+      // running on an Azure VM.
+      Assertions.assertThat(tokenProviderTypeName).describedAs("Token Provider Should be MsiTokenProvider").isInstanceOf(MsiTokenProvider.class);
+  }
+
   public void testGlobalAndAccountOAuthPrecedence(AbfsConfiguration abfsConf,
       AuthType globalAuthType,
       AuthType accountSpecificAuthType)