updated return type

Author: Amit-CloudSufi

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

@@ -20,6 +20,7 @@
 import io.cdap.cdap.api.exception.ErrorCodeType;
 import io.cdap.cdap.api.exception.ErrorType;
 import io.cdap.cdap.api.exception.ErrorUtils;
+import io.cdap.cdap.api.exception.ProgramFailureException;
 import io.cdap.plugin.snowflake.common.util.DocumentUrlUtil;
 import net.snowflake.client.jdbc.ErrorCode;
 
@@ -168,7 +169,8 @@ private static ErrorCategory getErrorCategoryFromSqlState(int errorCode) {
     return new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN);
   }
 
-  public static void fetchProgramFailureException(SQLException e, String errorReason, String errorMessage) {
+  public static ProgramFailureException fetchProgramFailureException(SQLException e, String errorReason,
+                                                                     String errorMessage) {
     Optional<ErrorCode> errorCodes = Arrays.stream(ErrorCode.values())
       .filter(errorCode -> errorCode.getSqlState().equals(e.getSQLState()))
       .findFirst();
@@ -177,7 +179,7 @@ public static void fetchProgramFailureException(SQLException e, String errorReas
       new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN);
     ErrorType errorType = errorCodes.isPresent() ? getErrorTypeFromErrorCode(errorCodes.get().getMessageCode()) :
       ErrorType.UNKNOWN;
-    throw ErrorUtils.getProgramFailureException(errorCategory, errorReason, errorMessage, errorType,
+    return ErrorUtils.getProgramFailureException(errorCategory, errorReason, errorMessage, errorType,
       true, ErrorCodeType.SQLSTATE, e.getSQLState(), DocumentUrlUtil.getSupportedDocumentUrl(), e);
   }
 }

src/main/java/io/cdap/plugin/snowflake/common/client/SnowflakeAccessor.java

@@ -70,7 +70,7 @@ public void runSQL(String query) {
         query, e.getSQLState(), e.getErrorCode(), e.getMessage());
       String errorReason = String.format("Statement '%s' failed with SQL state %s and error code %s. For more " +
         "details see %s.", query, e.getSQLState(), e.getErrorCode(), DocumentUrlUtil.getSupportedDocumentUrl());
-      SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
+      throw SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
     }
   }
 
@@ -100,7 +100,7 @@ public List<SnowflakeFieldDescriptor> describeQuery(String query) throws IOExcep
         "code %s with message: %s.", e.getSQLState(), e.getErrorCode(), e.getMessage());
       String errorReason = String.format("Failed to execute query to fetch descriptors with SQL State %s and error " +
         "code %s. For more details %s", e.getSQLState(), e.getErrorCode(), DocumentUrlUtil.getSupportedDocumentUrl());
-      SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
+      throw SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
     }
     return fieldDescriptors;
   }

src/main/java/io/cdap/plugin/snowflake/sink/batch/SnowflakeSinkAccessor.java

@@ -61,7 +61,7 @@ public void uploadStream(InputStream inputStream, String stageDir) {
       String errorMessage = String.format("Failed to compress '%s' and upload data to destination stage '%s' with " +
           "errorCode: '%s' and sqlState: '%s' with message: %s.", filename, stageDir, e.getErrorCode(), e.getSQLState(),
         e.getMessage());
-      SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
+      throw SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
     }
   }
 

src/main/java/io/cdap/plugin/snowflake/source/batch/SnowflakeSourceAccessor.java

@@ -98,7 +98,7 @@ public List<String> prepareStageSplits() {
         DocumentUrlUtil.getSupportedDocumentUrl());
       String errorMessage = String.format("Failed to load data into stage '%s' with sqlState %s and errorCode %s. " +
         "Failed to execute query with message: %s.", STAGE_PATH, e.getSQLState(), e.getErrorCode(), e.getMessage());
-      SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
+      throw SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
     }
     return stageSplits;
   }
@@ -130,8 +130,7 @@ public CSVReader buildCsvReader(String stageSplit) {
       String errorMessage = String.format("Failed to execute the query with sqlState: '%s' & errorCode: '%s' " +
           "with message: %s, stage split at %s.", e.getSQLState(), e.getErrorCode(),
         e.getMessage(), stageSplit);
-      SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
+      throw SnowflakeErrorType.fetchProgramFailureException(e, errorReason, errorMessage);
     }
-    return null;
   }
 }