Support GCP auth (#196)

## Changes
<!-- Summary of your changes that are easy to understand -->
- Make java-sdk support gcp auth

## Tests
<!-- How is this tested? -->
- Integration tests
- Unit tests
- Manual tests

Author: edwardfeng-db

NOTICE

@@ -30,6 +30,10 @@ This software contains code from the following open source projects, licensed un
 x/oauth2 - https://cs.opensource.google/go/x/oauth2/+/master:oauth2.go
 Copyright 2014 The Go Authors. All rights reserved.
 License - https://cs.opensource.google/go/x/oauth2/+/master:LICENSE
+
+googleapis/google-auth-library-java - https://github.com/googleapis/google-auth-library-java
+Copyright 2014, Google inc. All rights reserved.
+License - https://github.com/googleapis/google-auth-library-java/blob/main/LICENSE
 —--
 
 This Software contains code from the following open source projects, licensed under the MIT license:

README.md

@@ -173,6 +173,36 @@ import com.databricks.sdk.core.DatabricksConfig;
         WorkspaceClient workspace=new WorkspaceClient(config);
 ```
 
+### Google Cloud Platform native authentication
+
+By default, the Databricks SDK for Java first tries GCP credentials authentication (`auth_type='google-credentials'`, argument). If the SDK is unsuccessful, it then tries Google Cloud Platform (GCP) ID authentication (`auth_type='google-id'`, argument).
+
+The Databricks SDK for Java picks up an OAuth token in the scope of the Google Default Application Credentials (DAC) flow. This means that if you have run `gcloud auth application-default login` on your development machine, or launch the application on the compute, that is allowed to impersonate the Google Cloud service account specified in `google_service_account`. Authentication should then work out of the box. See [Creating and managing service accounts](https://cloud.google.com/iam/docs/creating-managing-service-accounts).
+
+To authenticate as a Google Cloud service account, you must provide one of the following:
+
+- `host` and `google_credentials`; or their environment variable or `.databrickscfg` file field equivalents.
+- `host` and `google_service_account`; or their environment variable or `.databrickscfg` file field equivalents.
+
+| Argument                 | Description | Environment variable |
+|--------------------------|-------------|--------------------------------------------------|
+| `google_credentials`     | _(String)_ GCP Service Account Credentials JSON or the location of these credentials on the local filesystem. | `GOOGLE_CREDENTIALS` |
+| `google_service_account` | _(String)_ The Google Cloud Platform (GCP) service account e-mail used for impersonation in the Default Application Credentials Flow that does not require a password. | `DATABRICKS_GOOGLE_SERVICE_ACCOUNT` |
+
+For example, to use Google ID authentication:
+
+```java
+import com.databricks.sdk.WorkspaceClient;
+import com.databricks.sdk.core.DatabricksConfig;
+...
+        DatabricksConfig config=new DatabricksConfig()
+        .setAuthType("google-credentials")
+        .setHost("https://my-databricks-instance.com")
+        .setGoogleServiceAccgount("google-service-account");
+        WorkspaceClient workspace=new WorkspaceClient(config);
+
+```
+
 ### Overriding `.databrickscfg`
 
 For [Databricks native authentication](#databricks-native-authentication), you can override the default behavior for using `.databrickscfg` as follows:

databricks-sdk-java/pom.xml

@@ -86,5 +86,11 @@
       <version>${mockito.version}</version>
       <scope>test</scope>
     </dependency>
+    <!-- Google Auth Library -->
+    <dependency>
+      <groupId>com.google.auth</groupId>
+      <artifactId>google-auth-library-oauth2-http</artifactId>
+      <version>1.20.0</version>
+    </dependency>
   </dependencies>
 </project>

databricks-sdk-java/src/main/java/com/databricks/sdk/core/DefaultCredentialsProvider.java

@@ -29,7 +29,9 @@ public DefaultCredentialsProvider() {
             new AzureCliCredentialsProvider(),
             new ExternalBrowserCredentialsProvider(),
             new DatabricksCliCredentialsProvider(),
-            new NotebookNativeCredentialsProvider());
+            new NotebookNativeCredentialsProvider(),
+            new GoogleCredentialsCredentialsProvider(),
+            new GoogleIdCredentialsProvider());
   }
 
   @Override

databricks-sdk-java/src/main/java/com/databricks/sdk/core/GoogleCredentialsCredentialsProvider.java

@@ -0,0 +1,83 @@
+package com.databricks.sdk.core;
+
+import static com.databricks.sdk.core.utils.GoogleUtils.GCP_SCOPES;
+import static com.databricks.sdk.core.utils.GoogleUtils.SA_ACCESS_TOKEN_HEADER;
+
+import com.google.auth.oauth2.*;
+import com.google.auth.oauth2.IdTokenProvider.Option;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GoogleCredentialsCredentialsProvider implements CredentialsProvider {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(GoogleCredentialsCredentialsProvider.class);
+
+  @Override
+  public String authType() {
+    return "google-credentials";
+  }
+
+  @Override
+  public HeaderFactory configure(DatabricksConfig config) {
+    String host = config.getHost();
+    String googleCredentials = config.getGoogleCredentials();
+    if (host == null || googleCredentials == null || !config.isGcp()) {
+      return null;
+    }
+
+    ServiceAccountCredentials serviceAccountCredentials;
+
+    try {
+      serviceAccountCredentials =
+          ServiceAccountCredentials.fromStream(Files.newInputStream(Paths.get(googleCredentials)));
+    } catch (IOException e) {
+      try {
+        // If file doesn't exist, we try to parse the config as the content.
+        serviceAccountCredentials =
+            ServiceAccountCredentials.fromStream(
+                new ByteArrayInputStream(googleCredentials.getBytes(StandardCharsets.UTF_8)));
+      } catch (IOException ex) {
+        LOG.warn("Failed to get Google service account credentials." + ex);
+        return null;
+      }
+    }
+
+    List<Option> tokenOption = Collections.emptyList();
+
+    ServiceAccountCredentials finalServiceAccountCredentials = serviceAccountCredentials;
+    return () -> {
+      IdToken idToken;
+      try {
+        idToken = finalServiceAccountCredentials.idTokenWithAudience(host, tokenOption);
+      } catch (IOException e) {
+        String message = "Failed to get id token from Google service account credentials.";
+        LOG.error(message + e);
+        throw new DatabricksException(message, e);
+      }
+      Map<String, String> headers = new HashMap<>();
+      headers.put("Authorization", String.format("Bearer %s", idToken.getTokenValue()));
+
+      if (config.isAccountClient()) {
+        AccessToken token;
+        try {
+          token = finalServiceAccountCredentials.createScoped(GCP_SCOPES).refreshAccessToken();
+        } catch (IOException e) {
+          String message =
+              "Failed to refresh access token from Google service account credentials.";
+          LOG.error(message + e);
+          throw new DatabricksException(message, e);
+        }
+        headers.put(SA_ACCESS_TOKEN_HEADER, token.getTokenValue());
+      }
+
+      return headers;
+    };
+  }
+}

databricks-sdk-java/src/main/java/com/databricks/sdk/core/GoogleIdCredentialsProvider.java

@@ -0,0 +1,84 @@
+package com.databricks.sdk.core;
+
+import static com.databricks.sdk.core.utils.GoogleUtils.GCP_SCOPES;
+import static com.databricks.sdk.core.utils.GoogleUtils.SA_ACCESS_TOKEN_HEADER;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.IdTokenCredentials;
+import com.google.auth.oauth2.IdTokenProvider;
+import com.google.auth.oauth2.ImpersonatedCredentials;
+import java.io.IOException;
+import java.util.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GoogleIdCredentialsProvider implements CredentialsProvider {
+
+  private static final Logger LOG = LoggerFactory.getLogger(GoogleIdCredentialsProvider.class);
+
+  @Override
+  public String authType() {
+    return "google-id";
+  }
+
+  @Override
+  public HeaderFactory configure(DatabricksConfig config) {
+    String host = config.getHost();
+    String googleServiceAccount = config.getGoogleServiceAccount();
+    if (host == null || googleServiceAccount == null || !config.isGcp()) {
+      return null;
+    }
+
+    GoogleCredentials googleCredentials;
+    try {
+      googleCredentials = GoogleCredentials.getApplicationDefault();
+    } catch (IOException e) {
+      LOG.warn("Failed to get Google application default credential." + e);
+      return null;
+    }
+
+    // Create the impersonated credential. Use 3600s as the lifetime, which is the default value in
+    // google-auth.
+    ImpersonatedCredentials impersonatedCredentials =
+        ImpersonatedCredentials.create(
+            googleCredentials, googleServiceAccount, null, new ArrayList<>(), 3600);
+
+    IdTokenCredentials idTokenCredentials =
+        IdTokenCredentials.newBuilder()
+            .setIdTokenProvider(impersonatedCredentials)
+            .setTargetAudience(host)
+            // Setting this will include email in the id token.
+            .setOptions(Collections.singletonList(IdTokenProvider.Option.INCLUDE_EMAIL))
+            .build();
+
+    ImpersonatedCredentials gcpScopedCredentials =
+        ImpersonatedCredentials.create(
+            googleCredentials, googleServiceAccount, null, GCP_SCOPES, 3600);
+
+    return () -> {
+      Map<String, String> headers = new HashMap<>();
+      try {
+        headers.put(
+            "Authorization",
+            String.format("Bearer %s", idTokenCredentials.refreshAccessToken().getTokenValue()));
+      } catch (IOException e) {
+        String message = "Failed to refresh access token from id token credentials.";
+        LOG.error(message + e);
+        throw new DatabricksException(message, e);
+      }
+
+      if (config.isAccountClient()) {
+        try {
+          headers.put(
+              SA_ACCESS_TOKEN_HEADER, gcpScopedCredentials.refreshAccessToken().getTokenValue());
+        } catch (IOException e) {
+          String message = "Failed to refresh access token from scoped id token credentials.";
+          LOG.error(message + e);
+          throw new DatabricksException(message, e);
+        }
+      }
+
+      return headers;
+    };
+  }
+}

databricks-sdk-java/src/main/java/com/databricks/sdk/core/utils/GoogleUtils.java

@@ -0,0 +1,13 @@
+package com.databricks.sdk.core.utils;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class GoogleUtils {
+  public static List<String> GCP_SCOPES =
+      Arrays.asList(
+          "https://www.googleapis.com/auth/cloud-platform",
+          "https://www.googleapis.com/auth/compute");
+
+  public static String SA_ACCESS_TOKEN_HEADER = "X-Databricks-GCP-SA-Access-Token";
+}

databricks-sdk-java/src/test/java/com/databricks/sdk/integration/CredentialsIT.java

@@ -15,10 +15,13 @@
 public class CredentialsIT {
   @Test
   void lists(AccountClient a) {
-    Iterable<Credential> list = a.credentials().list();
+    // Skipping this test for GCP because this api is not enabled in GCP.
+    if (!a.config().isGcp()) {
+      Iterable<Credential> list = a.credentials().list();
 
-    java.util.List<Credential> all = CollectionUtils.asList(list);
+      java.util.List<Credential> all = CollectionUtils.asList(list);
 
-    CollectionUtils.assertUnique(all);
+      CollectionUtils.assertUnique(all);
+    }
   }
 }

databricks-sdk-java/src/test/java/com/databricks/sdk/integration/FilesIT.java

@@ -5,7 +5,6 @@
 import com.databricks.sdk.integration.framework.EnvTest;
 import com.databricks.sdk.integration.framework.NameUtils;
 import com.databricks.sdk.integration.framework.ResourceWithCleanup;
-import com.databricks.sdk.service.files.FileInfo;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -46,10 +45,6 @@ private void writeFileAndReadFileInner(
     // Write the file to DBFS.
     workspace.files().upload(fileName, inputStream);
 
-    // Read file status
-    FileInfo fileStatus = workspace.files().getStatus(fileName);
-    Assertions.assertEquals(fileStatus.getPath(), fileName);
-
     // Read the file back from DBFS.
     try (InputStream readContents = workspace.files().download(fileName).getContents()) {
       byte[] result = new byte[fileContents.length];