addressed comments

Author: AnkitCLI

src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java

@@ -91,6 +91,8 @@ public class GCPUtils {
   public static final int BQ_DEFAULT_READ_TIMEOUT_SECONDS = 120;
   public static final String DATASTORE_SUPPORTED_DOC_URL = "https://cloud.google.com/datastore/docs/concepts/errors";
   public static final String BIG_TABLE_SUPPORTED_DOC_URL = "https://cloud.google.com/bigtable/docs/status-codes";
+  public static final String GCE_METADATA_SERVER_ERROR_SUPPORTED_DOC_URL =
+    "https://cloud.google.com/compute/docs/troubleshooting/troubleshoot-metadata-server";
 
   /**
    * Load a service account from the local file system.

src/main/java/io/cdap/plugin/gcp/common/ServerErrorException.java

@@ -31,9 +31,10 @@ public class ServerErrorException extends RuntimeException {
    *
    * @param statusCode the HTTP status code (should be in the 5xx range)
    * @param message    the detail message explaining the error
+   * @param cause      the original cause of the error
    */
-  public ServerErrorException(int statusCode, String message) {
-    super("Server error [" + statusCode + "]: " + message);
+  public ServerErrorException(int statusCode, String message, Throwable cause) {
+    super("Server error [" + statusCode + "]: " + message, cause);
     this.statusCode = statusCode;
   }
 

src/main/java/io/cdap/plugin/gcp/gcs/ServiceAccountAccessTokenProvider.java

@@ -22,13 +22,10 @@
 import com.google.cloud.hadoop.util.AccessTokenProvider;
 import com.google.cloud.hadoop.util.CredentialFactory;
 import dev.failsafe.Failsafe;
+import dev.failsafe.FailsafeException;
 import dev.failsafe.RetryPolicy;
-import io.cdap.cdap.api.exception.ErrorCategory;
-import io.cdap.cdap.api.exception.ErrorCategory.ErrorCategoryEnum;
 import io.cdap.cdap.api.exception.ErrorType;
-import io.cdap.cdap.api.exception.ErrorUtils;
-import io.cdap.plugin.gcp.bigquery.source.BigQuerySourceConfig;
-import io.cdap.plugin.gcp.bigquery.util.BigQueryConstants;
+import io.cdap.plugin.gcp.common.GCPErrorDetailsProviderUtil;
 import io.cdap.plugin.gcp.common.GCPUtils;
 import io.cdap.plugin.gcp.common.ServerErrorException;
 import org.apache.hadoop.conf.Configuration;
@@ -40,6 +37,7 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Date;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -52,68 +50,61 @@ public class ServiceAccountAccessTokenProvider implements AccessTokenProvider {
   private Configuration conf;
   private GoogleCredentials credentials;
   private static final Gson GSON = new Gson();
-  private static final Logger logger = LoggerFactory.getLogger(ServiceAccountAccessTokenProvider.class);
+  private static final Logger LOG = LoggerFactory.getLogger(ServiceAccountAccessTokenProvider.class);
   public static final int DEFAULT_INITIAL_RETRY_DURATION_SECONDS = 5;
   public static final int DEFAULT_MAX_RETRY_COUNT = 5;
   public static final int DEFAULT_MAX_RETRY_DURATION_SECONDS = 80;
-
+  private static final RetryPolicy<Object> RETRY_POLICY = createRetryPolicy();
+  private static final Pattern SERVER_ERROR_PATTERN = Pattern.compile("Unexpected Error code 5\\d{2} trying to get " +
+  "security access token from Compute Engine metadata for the default service account.*");
 
   @Override
   public AccessToken getAccessToken() {
-    int initialRetryDuration = DEFAULT_INITIAL_RETRY_DURATION_SECONDS;
-    int maxRetryCount = DEFAULT_MAX_RETRY_COUNT;
-    int maxRetryDuration =  DEFAULT_MAX_RETRY_DURATION_SECONDS;
-      logger.debug(
-        "Initializing RetryPolicy with the following configuration: MaxRetryCount: {}, InitialRetryDuration: {}s, " +
-          "MaxRetryDuration: {}s", maxRetryCount, initialRetryDuration, maxRetryDuration);
       try {
-        return Failsafe.with(getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount))
-          .get(() -> {
-            com.google.auth.oauth2.AccessToken token = safeGetAccessToken();
+        return Failsafe.with(RETRY_POLICY).get(() -> {
+            com.google.auth.oauth2.AccessToken token = retrieveAccessToken();
             if (token == null || token.getExpirationTime().before(Date.from(Instant.now()))) {
               refresh();
-              token = safeGetAccessToken();
+              token = retrieveAccessToken();
             }
             return new AccessToken(token.getTokenValue(), token.getExpirationTime().getTime());
           });
-      } catch (Exception e) {
-        throw ErrorUtils.getProgramFailureException(
-          new ErrorCategory(ErrorCategoryEnum.PLUGIN),
-          "Unable to get service account access token after retries.",
-          e.getMessage(),
-          ErrorType.UNKNOWN,
-          true,
-          e
+      } catch (FailsafeException e) {
+        Throwable t = e.getCause() != null ? e.getCause() : e;
+        ErrorType errorType = (t instanceof ServerErrorException) ? ErrorType.SYSTEM : ErrorType.UNKNOWN;
+        throw GCPErrorDetailsProviderUtil.getHttpResponseExceptionDetailsFromChain(
+          e, "Unable to get service account access token after retries.", errorType, true,
+          GCPUtils.GCE_METADATA_SERVER_ERROR_SUPPORTED_DOC_URL
         );
       }
     }
 
-
-  private RetryPolicy<Object> getRetryPolicy(int initialRetryDuration, int maxRetryDuration,
-                                             int maxRetryCount) {
+  private static RetryPolicy<Object> createRetryPolicy() {
     return RetryPolicy.builder()
       .handle(ServerErrorException.class)
-      .withBackoff(Duration.ofSeconds(initialRetryDuration), Duration.ofSeconds(maxRetryDuration))
-      .withMaxRetries(maxRetryCount)
-      .onRetry(event -> logger.debug("Retry attempt {} due to {}", event.getAttemptCount(), event.getLastException().
+      .withBackoff(Duration.ofSeconds(DEFAULT_INITIAL_RETRY_DURATION_SECONDS),
+                   Duration.ofSeconds(DEFAULT_MAX_RETRY_DURATION_SECONDS))
+      .withMaxRetries(DEFAULT_MAX_RETRY_COUNT)
+      .onRetry(event -> LOG.debug("Retry attempt {} due to {}", event.getAttemptCount(), event.getLastException().
         getMessage()))
-      .onSuccess(event -> logger.debug("Access Token Fetched Successfully."))
-      .onRetriesExceeded(event -> logger.error("Retry limit reached for Service account."))
+      .onSuccess(event -> LOG.debug("Access Token Fetched Successfully."))
+      .onRetriesExceeded(
+       event -> LOG.error("Unable to get service account access token after {} retries.", event.getAttemptCount() - 1))
       .build();
   }
 
   private boolean isServerError(IOException e) {
     String msg = e.getMessage();
-    return msg != null && msg.matches("^5\\d{2}$"); // crude check for 5xx codes
+    return msg != null && SERVER_ERROR_PATTERN.matcher(msg).matches();
   }
 
-  private com.google.auth.oauth2.AccessToken safeGetAccessToken() throws IOException {
+  private com.google.auth.oauth2.AccessToken retrieveAccessToken() throws IOException {
     try {
       return getCredentials().getAccessToken();
     } catch (IOException e) {
       if (isServerError(e)) {
         throw new ServerErrorException(HttpStatus.SC_SERVICE_UNAVAILABLE, "Server error while fetching access token: "
-          + e.getMessage());
+          + e.getMessage(), e);
       }
       throw e;
     }
@@ -126,11 +117,11 @@ public void refresh() throws IOException {
     } catch (IOException e) {
       if (isServerError(e)) {
         throw new ServerErrorException(HttpStatus.SC_SERVICE_UNAVAILABLE, "Server error during refresh: " +
-          e.getMessage());
+          e.getMessage(), e);
       }
-      throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.PLUGIN),
-        "Unable to refresh service account access token.", e.getMessage(),
-        ErrorType.UNKNOWN, true, e);
+      throw GCPErrorDetailsProviderUtil.getHttpResponseExceptionDetailsFromChain(
+        e, "Unable to refresh service account access token.", ErrorType.UNKNOWN, true,
+        GCPUtils.GCE_METADATA_SERVER_ERROR_SUPPORTED_DOC_URL);
     }
   }