[PLUGIN-1824] ErrorDetailsProvider - MySql Source/Sink plugin

database-commons/src/main/java/io/cdap/plugin/db/DBErrorDetailsProvider.java

@@ -0,0 +1,117 @@
+/*
+ * 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.db;
+
+import com.google.common.base.Throwables;
+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 java.sql.SQLException;
+import java.util.List;
+
+/**
+ * A custom ErrorDetailsProvider for Database plugins.
+ */
+public class DBErrorDetailsProvider implements ErrorDetailsProvider {
+
+  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 SQLException) {
+        return getProgramFailureException((SQLException) t, errorContext);
+      }
+      if (t instanceof IllegalArgumentException) {
+        return getProgramFailureException((IllegalArgumentException) t, errorContext);
+      }
+      if (t instanceof IllegalStateException) {
+        return getProgramFailureException((IllegalStateException) t, errorContext);
+      }
+
+    }
+    return null;
+  }
+
+  /**
+   * Get a ProgramFailureException with the given error
+   * information from {@link SQLException}.
+   *
+   * @param e The SQLException to get the error information from.
+   * @return A ProgramFailureException with the given error information.
+   */
+  private ProgramFailureException getProgramFailureException(SQLException e, ErrorContext errorContext) {
+    String errorMessage = e.getMessage();
+    String sqlState = e.getSQLState();
+    int errorCode = e.getErrorCode();
+    String errorMessageWithDetails = String.format(
+      "Error occurred in the phase: '%s'. Error message: '%s'. Error code: '%s'. sqlState: '%s'",
+      errorContext.getPhase(), errorMessage, errorCode, sqlState);
+    return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+      errorMessage, errorMessageWithDetails, getErrorTypeFromErrorCode(errorCode), false, e);
+  }
+
+  /**
+   * 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 the external documentation link for the client errors if available.
+   *
+   * @return The external documentation link as a {@link String}.
+   */
+  protected String getExternalDocumentationLink() {
+    return null;
+  }
+
+  protected ErrorType getErrorTypeFromErrorCode(int errorCode) {
+    return ErrorType.UNKNOWN;
+  }
+}

database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java

@@ -21,6 +21,9 @@
 import io.cdap.cdap.api.common.Bytes;
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
+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.util.DBUtils;
 import io.cdap.plugin.util.Lazy;
 import org.apache.hadoop.conf.Configurable;
@@ -305,7 +308,10 @@ protected void updateOperation(PreparedStatement stmt) throws SQLException {
    * @throws SQLException
    */
   protected void upsertOperation(PreparedStatement stmt) throws SQLException {
-    throw new UnsupportedOperationException();
+    String errorMessage = "Upsert operation is not supported for this plugin.";
+    throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+      errorMessage, errorMessage, ErrorType.SYSTEM, false, new UnsupportedOperationException(errorMessage));
+
   }
 
   private boolean fillUpdateParams(List<String> updatedKeyList, ColumnType columnType) {
@@ -366,7 +372,9 @@ private void writeToDataOut(DataOutput out, Schema.Field field) throws IOExcepti
         out.write((byte[]) fieldValue);
         break;
       default:
-        throw new IOException(String.format("Unsupported datatype: %s with value: %s.", fieldType, fieldValue));
+        String errorMessage = String.format("Unsupported datatype: %s with value: %s.", fieldType, fieldValue);
+        throw  ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+          errorMessage, errorMessage, ErrorType.USER, false, new IOException(errorMessage));
     }
   }
 

database-commons/src/main/java/io/cdap/plugin/db/sink/AbstractDBSink.java

@@ -32,6 +32,7 @@
 import io.cdap.cdap.etl.api.StageConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
 import io.cdap.cdap.etl.api.batch.BatchSinkContext;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
 import io.cdap.cdap.etl.api.validation.InvalidStageException;
 import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.common.ReferenceBatchSink;
@@ -42,6 +43,7 @@
 import io.cdap.plugin.db.ConnectionConfig;
 import io.cdap.plugin.db.ConnectionConfigAccessor;
 import io.cdap.plugin.db.DBConfig;
+import io.cdap.plugin.db.DBErrorDetailsProvider;
 import io.cdap.plugin.db.DBRecord;
 import io.cdap.plugin.db.Operation;
 import io.cdap.plugin.db.SchemaReader;
@@ -163,6 +165,16 @@ public void validateOperations(FailureCollector collector, T dbSinkConfig, @Null
     }
   }
 
+  /**
+   * Returns the ErrorDetailsProvider class name.
+   * Override this method to provide a custom ErrorDetailsProvider class name.
+   *
+   * @return ErrorDetailsProvider class name
+   */
+  protected String getErrorDetailsProviderClassName() {
+    return DBErrorDetailsProvider.class.getName();
+  }
+
   @Override
   public void prepareRun(BatchSinkContext context) {
     String connectionString = dbSinkConfig.getConnectionString();
@@ -229,6 +241,9 @@ public void prepareRun(BatchSinkContext context) {
     }
 
     addOutputContext(context);
+
+    // set error details provider
+    context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(getErrorDetailsProviderClassName()));
   }
   protected void addOutputContext(BatchSinkContext context) {
     context.addOutput(Output.of(dbSinkConfig.getReferenceName(),

database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java

@@ -25,13 +25,17 @@
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.api.dataset.lib.KeyValue;
+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.plugin.PluginConfig;
 import io.cdap.cdap.etl.api.Emitter;
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.StageConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
 import io.cdap.cdap.etl.api.batch.BatchSourceContext;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
 import io.cdap.cdap.internal.io.SchemaTypeAdapter;
 import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.common.ReferenceBatchSource;
@@ -41,6 +45,7 @@
 import io.cdap.plugin.db.ConnectionConfig;
 import io.cdap.plugin.db.ConnectionConfigAccessor;
 import io.cdap.plugin.db.DBConfig;
+import io.cdap.plugin.db.DBErrorDetailsProvider;
 import io.cdap.plugin.db.DBRecord;
 import io.cdap.plugin.db.SchemaReader;
 import io.cdap.plugin.db.TransactionIsolationLevel;
@@ -119,8 +124,9 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
         collector.addFailure("Unable to instantiate JDBC driver: " + e.getMessage(), null)
           .withStacktrace(e.getStackTrace());
       } catch (SQLException e) {
-        collector.addFailure("SQL error while getting query schema: " + e.getMessage(), null)
-          .withStacktrace(e.getStackTrace());
+        String details = String.format("SQL error while getting query schema: Error: %s, SQLState: %s, ErrorCode: %s",
+          e.getMessage(), e.getSQLState(), e.getErrorCode());
+        collector.addFailure(details, null).withStacktrace(e.getStackTrace());
       } catch (Exception e) {
         collector.addFailure(e.getMessage(), null).withStacktrace(e.getStackTrace());
       }
@@ -194,7 +200,11 @@ private Schema loadSchemaFromDB(Class<? extends Driver> driverClass)
 
     } catch (SQLException e) {
       // wrap exception to ensure SQLException-child instances not exposed to contexts without jdbc driver in classpath
-      throw new SQLException(e.getMessage(), e.getSQLState(), e.getErrorCode());
+      String errorMessageWithDetails = String.format("Error occurred while trying to get schema from database." +
+        "Error message: '%s'. Error code: '%s'. SQLState: '%s'", e.getMessage(), e.getErrorCode(), e.getSQLState());
+      throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+        e.getMessage(), errorMessageWithDetails, ErrorType.USER, false, new SQLException(e.getMessage(),
+          e.getSQLState(), e.getErrorCode()));
     } finally {
       driverCleanup.destroy();
     }
@@ -212,6 +222,16 @@ protected SchemaReader getSchemaReader() {
     return new CommonSchemaReader();
   }
 
+  /**
+   * Returns the ErrorDetailsProvider class name.
+   * Override this method to provide a custom ErrorDetailsProvider class name.
+   *
+   * @return ErrorDetailsProvider class name
+   */
+  protected String getErrorDetailsProviderClassName() {
+    return DBErrorDetailsProvider.class.getName();
+  }
+
   private DriverCleanup loadPluginClassAndGetDriver(Class<? extends Driver> driverClass)
     throws IllegalAccessException, InstantiationException, SQLException {
 
@@ -270,6 +290,9 @@ public void prepareRun(BatchSourceContext context) throws Exception {
     }
     context.setInput(Input.of(sourceConfig.getReferenceName(), new SourceInputFormatProvider(
       DataDrivenETLDBInputFormat.class, connectionConfigAccessor.getConfiguration())));
+
+    // set error details provider
+    context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(getErrorDetailsProviderClassName()));
   }
 
   /**

mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConstants.java

@@ -16,12 +16,18 @@
 
 package io.cdap.plugin.mysql;
 
+import io.cdap.cdap.api.exception.ErrorCategory;
+import io.cdap.cdap.api.exception.ErrorType;
+import io.cdap.cdap.api.exception.ErrorUtils;
+
 /**
  * MySQL Constants.
  */
 public final class MysqlConstants {
   private MysqlConstants() {
-    throw new AssertionError("Should not instantiate static utility class.");
+    String errorMessage = "Should not instantiate static utility class.";
+    throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+      errorMessage, errorMessage, ErrorType.SYSTEM, false, new AssertionError(errorMessage));
   }
 
   public static final String PLUGIN_NAME = "Mysql";

mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlErrorDetailsProvider.java

@@ -0,0 +1,44 @@
+/*
+ * 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.mysql;
+
+import io.cdap.cdap.api.exception.ErrorType;
+import io.cdap.plugin.db.DBErrorDetailsProvider;
+
+/**
+ * A custom ErrorDetailsProvider for MySQL plugins.
+ */
+public class MysqlErrorDetailsProvider extends DBErrorDetailsProvider {
+
+  @Override
+  protected String getExternalDocumentationLink() {
+    return "https://dev.mysql.com/doc/mysql-errors/9.0/en/";
+  }
+
+  @Override
+  protected ErrorType getErrorTypeFromErrorCode(int errorCode) {
+    // https://dev.mysql.com/doc/refman/9.0/en/error-message-elements.html#error-code-ranges
+    if (errorCode >= 1000 && errorCode <= 1999 || errorCode >= 2000 && errorCode <= 2999 ||
+      errorCode >= 3000 && errorCode <= 4999 || errorCode >= 5000 && errorCode <= 5999) {
+      return ErrorType.USER;
+    } else if (errorCode >= 10000 && errorCode <= 49999 || errorCode >= 50000 && errorCode <= 51999) {
+      return ErrorType.SYSTEM;
+    } else {
+      return ErrorType.UNKNOWN;
+    }
+  }
+}

mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSink.java

@@ -109,6 +109,11 @@ String getDbColumns() {
     return dbColumns;
   }
 
+  @Override
+  protected String getErrorDetailsProviderClassName() {
+    return MysqlErrorDetailsProvider.class.getName();
+  }
+
   /**
    * MySQL action configuration.
    */

mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSource.java

@@ -84,6 +84,11 @@ protected SchemaReader getSchemaReader() {
     return new MysqlSchemaReader(null, mysqlSourceConfig.getConnectionArguments());
   }
 
+  @Override
+  protected String getErrorDetailsProviderClassName() {
+    return MysqlErrorDetailsProvider.class.getName();
+  }
+
   /**
    * MySQL source config.
    */

mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlUtil.java

@@ -17,6 +17,9 @@
 package io.cdap.plugin.mysql;
 
 import com.google.common.collect.ImmutableMap;
+import io.cdap.cdap.api.exception.ErrorCategory;
+import io.cdap.cdap.api.exception.ErrorType;
+import io.cdap.cdap.api.exception.ErrorUtils;
 
 import java.sql.Types;
 import java.util.Map;
@@ -26,7 +29,9 @@
  */
 public final class MysqlUtil {
   private MysqlUtil() {
-    throw new AssertionError("Should not instantiate static utility class.");
+    String errorMessage = "Should not instantiate static utility class.";
+    throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+      errorMessage, errorMessage, ErrorType.SYSTEM, false, new AssertionError(errorMessage));
   }
 
   /**

oracle-plugin/src/e2e-test/resources/errorMessage.properties

@@ -7,7 +7,7 @@ errorMessageInvalidNumberOfSplits=Invalid value for Number of Splits '0'. Must b
 errorMessageNumberOfSplitNotNumber=Unable to create config for batchsource Oracle 'numSplits' is invalid: Value of field\
   \ class io.cdap.plugin.db.config.AbstractDBSpecificSourceConfig.numSplits is expected to be a number.
 errorMessageInvalidFetchSize=Invalid fetch size. Fetch size must be a positive integer.
-errorMessageInvalidSourceDatabase=SQL error while getting query schema: Listener refused the connection with the following \
+errorMessageInvalidSourceDatabase=SQL error while getting query schema: Error: Listener refused the connection with the following \
   error: ORA-12514, TNS:listener does not currently know of service requested in connect descriptor
 errorMessageInvalidImportQuery=Import Query select must contain the string '$CONDITIONS'. if Number of Splits is not set\
   \ to 1. Include '$CONDITIONS' in the Import Query