Support GCS Connector usage as a Hadoop Credential Provider. (#882)

For additional context, see
https://issues.apache.org/jira/browse/HADOOP-12846.

Closes #219.

gcs/CHANGES.md

@@ -79,6 +79,9 @@
     timeout is enforced during TLS handshakes when using Conscrypt as the
     security provider.
 
+1.  The Google Cloud Storage Connector now can be used as a
+    [Hadoop Credential Provider](https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/CredentialProviderAPI.html).
+
 ### 2.2.2 - 2021-06-25
 
 1.  Support footer prefetch in gRPC read channel.

gcs/src/main/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystem.java

@@ -114,6 +114,7 @@
 import org.apache.hadoop.fs.statistics.IOStatistics;
 import org.apache.hadoop.fs.statistics.IOStatisticsSource;
 import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.ProviderUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.LambdaUtils;
@@ -235,6 +236,8 @@ public void initialize(URI path, Configuration config) throws IOException {
     checkArgument(path.getScheme() != null, "scheme of path must not be null");
     checkArgument(path.getScheme().equals(getScheme()), "URI scheme not supported: %s", path);
 
+    config =
+        ProviderUtils.excludeIncompatibleCredentialProviders(config, GoogleHadoopFileSystem.class);
     super.initialize(path, config);
 
     initUri = path;

gcs/src/test/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemIntegrationTest.java

@@ -47,6 +47,7 @@
 import static com.google.common.truth.Truth.assertWithMessage;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Arrays.stream;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_CREDENTIAL_PROVIDER_PATH;
 import static org.junit.Assert.assertThrows;
 
 import com.google.api.client.http.HttpResponseException;
@@ -99,6 +100,8 @@
 @RunWith(JUnit4.class)
 public class GoogleHadoopFileSystemIntegrationTest extends GoogleHadoopFileSystemTestBase {
 
+  private static final String PUBLIC_BUCKET = "gs://gcp-public-data-landsat";
+
   @ClassRule
   public static NotInheritableExternalResource storageResource =
       new NotInheritableExternalResource(GoogleHadoopFileSystemIntegrationTest.class) {
@@ -1793,14 +1796,12 @@ public void testImpersonationInvalidUserNameIdentifierUsed() throws Exception {
 
   @Test
   public void unauthenticatedAccessToPublicBuckets_fsGsProperties() throws Exception {
-    String publicBucket = "gs://gcp-public-data-landsat";
-
     Configuration config = new Configuration();
     config.setEnum("fs.gs.auth.type", AuthenticationType.UNAUTHENTICATED);
 
-    FileSystem fs = FileSystem.get(new URI(publicBucket), config);
+    FileSystem fs = FileSystem.get(new URI(PUBLIC_BUCKET), config);
 
-    FileStatus[] fileStatuses = fs.listStatus(new Path(publicBucket));
+    FileStatus[] fileStatuses = fs.listStatus(new Path(PUBLIC_BUCKET));
 
     assertThat(
             ((GoogleHadoopFileSystem) fs)
@@ -1820,15 +1821,24 @@ public void unauthenticatedAccessToPublicBuckets_fsGsProperties() throws Excepti
 
   @Test
   public void unauthenticatedAccessToPublicBuckets_googleCloudProperties() throws Exception {
-    String publicBucket = "gs://gcp-public-data-landsat";
-
     Configuration config = new Configuration();
     config.setEnum("google.cloud.auth.type", AuthenticationType.UNAUTHENTICATED);
 
-    FileSystem fs = FileSystem.get(new URI(publicBucket), config);
+    FileSystem fs = FileSystem.get(new URI(PUBLIC_BUCKET), config);
 
-    FileStatus[] fileStatuses = fs.listStatus(new Path(publicBucket));
+    FileStatus[] fileStatuses = fs.listStatus(new Path(PUBLIC_BUCKET));
 
     assertThat(fileStatuses).isNotEmpty();
   }
+
+  @Test
+  public void testInitializeCompatibleWithHadoopCredentialProvider() throws Exception {
+    Configuration config = loadConfig();
+
+    // This does not need to refer to a real bucket/path for the test.
+    config.set(HADOOP_SECURITY_CREDENTIAL_PROVIDER_PATH, "jceks://gs@foobar/test.jceks");
+
+    FileSystem.get(new URI(PUBLIC_BUCKET), config);
+    // Initialization successful with no exception thrown.
+  }
 }

gcs/src/test/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemTest.java

@@ -262,4 +262,7 @@ public void unauthenticatedAccessToPublicBuckets_fsGsProperties() {}
 
   @Override
   public void unauthenticatedAccessToPublicBuckets_googleCloudProperties() {}
+
+  @Override
+  public void testInitializeCompatibleWithHadoopCredentialProvider() {}
 }