Merge pull request #43 from cloudsufi/snowFlakeError

[PLUGIN-1832] Error Management Snowflake Source and Sink, fix sonar issues, and added new Validation and fix bugs for maximum split size and NPE issue handled

Author: psainics

src/main/java/io/cdap/plugin/snowflake/common/BaseSnowflakeConfig.java

@@ -230,33 +230,10 @@ public String getConnectionArguments() {
   }
 
   public void validate(FailureCollector collector) {
-    if (getOauth2Enabled()) {
-      if (!containsMacro(PROPERTY_CLIENT_ID)
-        && Strings.isNullOrEmpty(getClientId())) {
-        collector.addFailure("Client ID is not set.", null)
-          .withConfigProperty(PROPERTY_CLIENT_ID);
-      }
-      if (!containsMacro(PROPERTY_CLIENT_SECRET)
-        && Strings.isNullOrEmpty(getClientSecret())) {
-        collector.addFailure("Client Secret is not set.", null)
-          .withConfigProperty(PROPERTY_CLIENT_SECRET);
-      }
-      if (!containsMacro(PROPERTY_REFRESH_TOKEN)
-        && Strings.isNullOrEmpty(getRefreshToken())) {
-        collector.addFailure("Refresh Token is not set.", null)
-          .withConfigProperty(PROPERTY_REFRESH_TOKEN);
-      }
-    } else if (getKeyPairEnabled()) {
-      if (!containsMacro(PROPERTY_USERNAME)
-        && Strings.isNullOrEmpty(getUsername())) {
-        collector.addFailure("Username is not set.", null)
-          .withConfigProperty(PROPERTY_USERNAME);
-      }
-      if (!containsMacro(PROPERTY_PRIVATE_KEY)
-        && Strings.isNullOrEmpty(getPrivateKey())) {
-        collector.addFailure("Private Key is not set.", null)
-          .withConfigProperty(PROPERTY_PRIVATE_KEY);
-      }
+    if (Boolean.TRUE.equals(getOauth2Enabled())) {
+      validateWhenOath2Enabled(collector);
+    } else if (Boolean.TRUE.equals(getKeyPairEnabled())) {
+      validateWhenKeyPairEnabled(collector);
     } else {
       if (!containsMacro(PROPERTY_USERNAME)
         && Strings.isNullOrEmpty(getUsername())) {
@@ -272,6 +249,37 @@ public void validate(FailureCollector collector) {
     validateConnection(collector);
   }
 
+  private void validateWhenKeyPairEnabled(FailureCollector collector) {
+    if (!containsMacro(PROPERTY_USERNAME)
+      && Strings.isNullOrEmpty(getUsername())) {
+      collector.addFailure("Username is not set.", null)
+        .withConfigProperty(PROPERTY_USERNAME);
+    }
+    if (!containsMacro(PROPERTY_PRIVATE_KEY)
+      && Strings.isNullOrEmpty(getPrivateKey())) {
+      collector.addFailure("Private Key is not set.", null)
+        .withConfigProperty(PROPERTY_PRIVATE_KEY);
+    }
+  }
+
+  private void validateWhenOath2Enabled(FailureCollector collector) {
+    if (!containsMacro(PROPERTY_CLIENT_ID)
+      && Strings.isNullOrEmpty(getClientId())) {
+      collector.addFailure("Client ID is not set.", null)
+        .withConfigProperty(PROPERTY_CLIENT_ID);
+    }
+    if (!containsMacro(PROPERTY_CLIENT_SECRET)
+      && Strings.isNullOrEmpty(getClientSecret())) {
+      collector.addFailure("Client Secret is not set.", null)
+        .withConfigProperty(PROPERTY_CLIENT_SECRET);
+    }
+    if (!containsMacro(PROPERTY_REFRESH_TOKEN)
+      && Strings.isNullOrEmpty(getRefreshToken())) {
+      collector.addFailure("Refresh Token is not set.", null)
+        .withConfigProperty(PROPERTY_REFRESH_TOKEN);
+    }
+  }
+
   public boolean canConnect() {
     return (!containsMacro(PROPERTY_DATABASE) && !containsMacro(PROPERTY_SCHEMA_NAME)
       && !containsMacro(PROPERTY_ACCOUNT_NAME) && !containsMacro(PROPERTY_USERNAME)
@@ -299,7 +307,7 @@ protected void validateConnection(FailureCollector collector) {
         .withConfigProperty(PROPERTY_USERNAME);
 
       // TODO: for oauth2
-      if (keyPairEnabled) {
+      if (Boolean.TRUE.equals(keyPairEnabled)) {
         failure.withConfigProperty(PROPERTY_PRIVATE_KEY);
       } else {
         failure.withConfigProperty(PROPERTY_PASSWORD);

src/main/java/io/cdap/plugin/snowflake/common/OAuthUtil.java

@@ -19,13 +19,17 @@
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
 import com.google.gson.JsonSyntaxException;
+import io.cdap.cdap.api.exception.ErrorCategory;
+import io.cdap.cdap.api.exception.ErrorType;
+import io.cdap.cdap.api.exception.ErrorUtils;
 import io.cdap.plugin.snowflake.common.exception.ConnectionTimeoutException;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.util.EntityUtils;
+
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -50,16 +54,23 @@ public static String getAccessTokenByRefreshToken(CloseableHttpClient httpclient
       httppost.setHeader("Content-type", "application/x-www-form-urlencoded");
 
       // set grant type and refresh_token. It should be in body not url!
-      StringEntity entity = new StringEntity(String.format("refresh_token=%s&grant_type=refresh_token",
-                                                           URLEncoder.encode(config.getRefreshToken(), "UTF-8")));
-      httppost.setEntity(entity);
+      try {
+        StringEntity entity = new StringEntity(String.format("refresh_token=%s&grant_type=refresh_token",
+          URLEncoder.encode(config.getRefreshToken(), "UTF-8")));
+        httppost.setEntity(entity);
+      } catch (NullPointerException e) {
+        String errorMessage = String.format("Failed to encode URL due to missing Refresh Token with message: %s.",
+          e.getMessage());
+        String errorReason = "Error encoding URL due to missing Refresh Token.";
+        throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+          errorReason, errorMessage, ErrorType.USER, true, e);
+      }
 
       // set 'Authorization' header
       String stringToEncode = config.getClientId() + ":" + config.getClientSecret();
       String encondedAuthorization = new String(Base64.getEncoder().encode(stringToEncode.getBytes()));
       httppost.setHeader("Authorization", String.format("Basic %s", encondedAuthorization));
 
-
       CloseableHttpResponse response = httpclient.execute(httppost);
       String responseString = EntityUtils.toString(response.getEntity(), "UTF-8");
 
@@ -72,7 +83,13 @@ public static String getAccessTokenByRefreshToken(CloseableHttpClient httpclient
 
       // if exception happened during parsing OR if json does not contain 'access_token' key.
       if (jsonElement == null) {
-        throw new RuntimeException(String.format("Unexpected response '%s' from '%s'", responseString, uri.toString()));
+        String errorReason = String.format("Failed to parse access token from response. Request %s returned response " +
+            "code '%s' & reason: %s", uri.toString(), response.getStatusLine().getStatusCode(),
+          response.getStatusLine().getReasonPhrase());
+        String errorMessage = String.format("Failed to parse access token, request %s returned response %s " +
+          "with code '%s'.", uri, responseString, response.getStatusLine().getStatusCode());
+        throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+          errorReason, errorMessage, ErrorType.SYSTEM, true, new JsonSyntaxException(errorReason));
       }
 
       return jsonElement.getAsString();

src/main/java/io/cdap/plugin/snowflake/common/SnowflakeErrorDetailsProvider.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.snowflake.common;
+
+import com.google.common.base.Throwables;
+import io.cdap.cdap.api.data.format.UnexpectedFormatException;
+import io.cdap.cdap.api.exception.ErrorCategory;
+import io.cdap.cdap.api.exception.ErrorType;
+import io.cdap.cdap.api.exception.ErrorUtils;
+import io.cdap.cdap.api.exception.ProgramFailureException;
+import io.cdap.cdap.etl.api.exception.ErrorContext;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProvider;
+import io.cdap.plugin.snowflake.common.exception.ConnectionTimeoutException;
+import io.cdap.plugin.snowflake.common.exception.SchemaParseException;
+
+import java.util.List;
+
+/**
+ * Error details provided for the Snowflake
+ **/
+public class SnowflakeErrorDetailsProvider implements ErrorDetailsProvider {
+
+  private static final String ERROR_MESSAGE_FORMAT = "Error occurred in the phase: '%s'. Error message: %s";
+
+  @Override
+  public ProgramFailureException getExceptionDetails(Exception e, ErrorContext errorContext) {
+    List<Throwable> causalChain = Throwables.getCausalChain(e);
+    for (Throwable t : causalChain) {
+      if (t instanceof ProgramFailureException) {
+        // if causal chain already has program failure exception, return null to avoid double wrap.
+        return null;
+      }
+      if (t instanceof IllegalArgumentException) {
+        return getProgramFailureException((IllegalArgumentException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof IllegalStateException) {
+        return getProgramFailureException((IllegalStateException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof SchemaParseException) {
+        return getProgramFailureException((SchemaParseException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof UnexpectedFormatException) {
+        return getProgramFailureException((UnexpectedFormatException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof ConnectionTimeoutException) {
+        return getProgramFailureException((ConnectionTimeoutException) t, errorContext, ErrorType.SYSTEM);
+      }
+    }
+    return null;
+  }
+
+  /**
+   * Get a ProgramFailureException with the given error
+   * information from {@link Exception}.
+   *
+   * @param exception The Exception to get the error information from.
+   * @return A ProgramFailureException with the given error information.
+   */
+  private ProgramFailureException getProgramFailureException(Exception exception, ErrorContext errorContext,
+                                                             ErrorType errorType) {
+    String errorMessage = exception.getMessage();
+    return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+      errorMessage,
+      String.format(ERROR_MESSAGE_FORMAT, errorContext.getPhase(), errorMessage), errorType, false, exception);
+  }
+}