-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Error management for Snowflake source and sink, Added new validation …
…for maximum split size and NPE issue handled
- Loading branch information
1 parent
1d9cacd
commit 756fb99
Showing
16 changed files
with
495 additions
and
80 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
161 changes: 161 additions & 0 deletions
161
src/main/java/io/cdap/plugin/snowflake/common/SnowflakeErrorDetailsProvider.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,161 @@ | ||
| /* | ||
| * Copyright © 2024 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.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.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.net.URISyntaxException; | ||
| import java.util.List; | ||
|
|
||
|
|
||
| /** | ||
| * Error details provided for the Snowflake | ||
| **/ | ||
| public class SnowflakeErrorDetailsProvider implements ErrorDetailsProvider { | ||
|
|
||
| @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); | ||
| } | ||
| if (t instanceof IllegalStateException) { | ||
| return getProgramFailureException((IllegalStateException) t, errorContext); | ||
| } | ||
| if (t instanceof URISyntaxException) { | ||
| return getProgramFailureException((URISyntaxException) t, errorContext); | ||
| } | ||
| if (t instanceof SchemaParseException) { | ||
| return getProgramFailureException((SchemaParseException) t, errorContext); | ||
| } | ||
| if (t instanceof UnexpectedFormatException) { | ||
| return getProgramFailureException((UnexpectedFormatException) t, errorContext); | ||
| } | ||
| if (t instanceof ConnectionTimeoutException) { | ||
| return getProgramFailureException((ConnectionTimeoutException) t, errorContext); | ||
| } | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| /** | ||
| * Get a ProgramFailureException with the given error | ||
| * information from {@link IllegalArgumentException}. | ||
| * | ||
| * @param e The IllegalArgumentException to get the error information from. | ||
| * @return A ProgramFailureException with the given error information. | ||
| */ | ||
| private ProgramFailureException getProgramFailureException(IllegalArgumentException e, ErrorContext errorContext) { | ||
| String errorMessage = e.getMessage(); | ||
| String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s"; | ||
|
|
||
| return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), | ||
| errorMessage, | ||
| String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.USER, false, e); | ||
| } | ||
|
|
||
| /** | ||
| * Get a ProgramFailureException with the given error | ||
| * information from {@link IllegalStateException}. | ||
| * | ||
| * @param e The IllegalStateException to get the error information from. | ||
| * @return A ProgramFailureException with the given error information. | ||
| */ | ||
| private ProgramFailureException getProgramFailureException(IllegalStateException e, ErrorContext errorContext) { | ||
| String errorMessage = e.getMessage(); | ||
| String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s"; | ||
| return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), | ||
| errorMessage, | ||
| String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.SYSTEM, false, e); | ||
| } | ||
|
|
||
| /** | ||
| * Get a ProgramFailureException with the given error | ||
| * information from {@link URISyntaxException}. | ||
| * | ||
| * @param e The URISyntaxException to get the error information from. | ||
| * @return A ProgramFailureException with the given error information. | ||
| */ | ||
| private ProgramFailureException getProgramFailureException(URISyntaxException e, | ||
| ErrorContext errorContext) { | ||
| String errorMessage = e.getMessage(); | ||
| String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s"; | ||
| return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), | ||
| errorMessage, | ||
| String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.USER, false, e); | ||
| } | ||
|
|
||
| /** | ||
| * Get a ProgramFailureException with the given error | ||
| * information from {@link SchemaParseException}. | ||
| * | ||
| * @param e The SchemaParseException to get the error information from. | ||
| * @return A ProgramFailureException with the given error information. | ||
| */ | ||
| private ProgramFailureException getProgramFailureException(SchemaParseException e, ErrorContext errorContext) { | ||
| String errorMessage = e.getMessage(); | ||
| String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s"; | ||
| return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), | ||
| errorMessage, | ||
| String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.USER, false, e); | ||
| } | ||
|
|
||
| /** | ||
| * Get a ProgramFailureException with the given error | ||
| * information from {@link UnexpectedFormatException}. | ||
| * | ||
| * @param e The UnexpectedFormatException to get the error information from. | ||
| * @return A ProgramFailureException with the given error information. | ||
| */ | ||
| private ProgramFailureException getProgramFailureException(UnexpectedFormatException e, ErrorContext errorContext) { | ||
| String errorMessage = e.getMessage(); | ||
| String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s"; | ||
| return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), | ||
| errorMessage, | ||
| String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.USER, false, e); | ||
| } | ||
|
|
||
| /** | ||
| * Get a ProgramFailureException with the given error | ||
| * information from {@link ConnectionTimeoutException}. | ||
| * | ||
| * @param e The ConnectionTimeoutException to get the error information from. | ||
| * @return A ProgramFailureException with the given error information. | ||
| */ | ||
| private ProgramFailureException getProgramFailureException(ConnectionTimeoutException e, ErrorContext errorContext) { | ||
| String errorMessage = e.getMessage(); | ||
| String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s"; | ||
| return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), | ||
| errorMessage, | ||
| String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.SYSTEM, false, e); | ||
| } | ||
| } |
59 changes: 59 additions & 0 deletions
59
src/main/java/io/cdap/plugin/snowflake/common/SnowflakeErrorType.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| /* | ||
| * Copyright © 2024 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 io.cdap.cdap.api.exception.ErrorType; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.HashSet; | ||
| import java.util.Set; | ||
|
|
||
| /** | ||
| * Error Type provided based on the Snowflake error message code | ||
| * | ||
| **/ | ||
| public class SnowflakeErrorType { | ||
|
|
||
| //https://github.com/snowflakedb/snowflake-jdbc/blob/master/src/main/java/net/snowflake/client/jdbc/ErrorCode.java | ||
| private static final Set<Integer> USER_ERRORS = new HashSet<>(Arrays.asList( | ||
| 200004, 200006, 200007, 200008, 200009, 200010, 200011, 200012, 200014, | ||
| 200017, 200018, 200019, 200021, 200023, 200024, 200025, 200026, 200028, | ||
| 200029, 200030, 200031, 200032, 200033, 200034, 200035, 200036, 200037, | ||
| 200038, 200045, 200046, 200047, 200056 | ||
| )); | ||
|
|
||
| private static final Set<Integer> SYSTEM_ERRORS = new HashSet<>(Arrays.asList( | ||
| 200001, 200002, 200003, 200013, 200015, 200016, 200020, 200022, 200039, | ||
| 200040, 200044, 200061 | ||
| )); | ||
|
|
||
| /** | ||
| * Method to get the error type based on the error code. | ||
| * | ||
| * @param errorCode the error code to classify | ||
| * @return the corresponding ErrorType (USER, SYSTEM, UNKNOWN) | ||
| */ | ||
| public static ErrorType getErrorType(int errorCode) { | ||
| if (USER_ERRORS.contains(errorCode)) { | ||
| return ErrorType.USER; | ||
| } else if (SYSTEM_ERRORS.contains(errorCode)) { | ||
| return ErrorType.SYSTEM; | ||
| } else { | ||
| return ErrorType.UNKNOWN; | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.