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

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

@@ -0,0 +1,124 @@
+/*
+ * 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.Strings;
+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);
+    String externalDocumentationLink = getExternalDocumentationLink();
+    if (!Strings.isNullOrEmpty(externalDocumentationLink)) {
+      if (!errorMessageWithDetails.endsWith(".")) {
+        errorMessageWithDetails = errorMessageWithDetails + ".";
+      }
+      errorMessageWithDetails = String.format("%s For more details, see %s", errorMessageWithDetails,
+        externalDocumentationLink);
+    }
+    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();
@@ -227,7 +239,8 @@ public void prepareRun(BatchSinkContext context) {
       configuration.set(ETLDBOutputFormat.COMMIT_BATCH_SIZE,
                         context.getArguments().get(ETLDBOutputFormat.COMMIT_BATCH_SIZE));
     }
-
+    // set error details provider
+    context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(getErrorDetailsProviderClassName()));
     addOutputContext(context);
   }
   protected void addOutputContext(BatchSinkContext context) {

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 {
 
@@ -268,6 +288,8 @@ public void prepareRun(BatchSourceContext context) throws Exception {
       lineageRecorder.recordRead("Read", "Read from database plugin",
                                  schema.getFields().stream().map(Schema.Field::getName).collect(Collectors.toList()));
     }
+    // set error details provider
+    context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(getErrorDetailsProviderClassName()));
     context.setInput(Input.of(sourceConfig.getReferenceName(), new SourceInputFormatProvider(
       DataDrivenETLDBInputFormat.class, connectionConfigAccessor.getConfiguration())));
   }

mysql-plugin/src/e2e-test/features/mysqlsink/RunTimeWithMacros.feature

@@ -100,6 +100,7 @@ Feature: MySQL Sink - Run time scenarios (macro)
     Then Wait till pipeline is in running state
     Then Open and capture logs
     And Verify the pipeline status is "Failed"
+    And Close the pipeline logs
 
   @BQ_SOURCE_TEST @MYSQL_TARGET_TABLE @Mysql_Required
   Scenario: Verify that the pipeline fails when user provides invalid Credentials for connection with Macros
@@ -135,4 +136,6 @@ Feature: MySQL Sink - Run time scenarios (macro)
     And Enter runtime argument value "invalid.password" for key "password"
     And Run the Pipeline in Runtime with runtime arguments
     And Wait till pipeline is in running state
+    And Open and capture logs
     And Verify the pipeline status is "Failed"
+    And Close the pipeline logs

mysql-plugin/src/e2e-test/features/mysqlsource/RunTime.feature

@@ -143,6 +143,7 @@ Feature: MySQL Source - Run time scenarios
     Then Save the pipeline
     Then Preview and run the pipeline
     Then Wait till pipeline preview is in running state
+    Then Open and capture pipeline preview logs
     Then Verify the preview run status of pipeline in the logs is "failed"
 
   @MYSQL_SOURCE_TEST @MYSQL_TARGET_TEST @Mysql_Required

mysql-plugin/src/e2e-test/features/mysqlsource/RunTimeWithMacros.feature

@@ -201,7 +201,9 @@ Feature: MySQL Source - Run time scenarios (macro)
     And Enter runtime argument value "invalid.query" for key "importQuery"
     And Run the Pipeline in Runtime with runtime arguments
     And Wait till pipeline is in running state
+    And Open and capture logs
     And Verify the pipeline status is "Failed"
+    And Close the pipeline logs
 
   @MYSQL_SOURCE_TEST @MYSQL_TARGET_TEST @Mysql_Required
   Scenario: Verify that pipeline fails when user provides invalid Credentials for connection with Macros
@@ -241,4 +243,6 @@ Feature: MySQL Source - Run time scenarios (macro)
     And Enter runtime argument value "invalid.password" for key "Password"
     And Run the Pipeline in Runtime with runtime arguments
     And Wait till pipeline is in running state
+    And Open and capture logs
     And Verify the pipeline status is "Failed"
+    And Close the pipeline logs

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,45 @@
+/*
+ * 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 <= 5999) {
+      return ErrorType.USER;
+    } else if (errorCode >= 10000 && errorCode <= 51999) {
+      // SYSTEM errors: Enterprise and user-defined custom error messages
+      return ErrorType.SYSTEM;
+    } else {
+      // UNKNOWN errors: Anything outside defined range
+      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.
    */