PLUGIN-1823: Revert SQLException handling changes for validateSchema

Author: sgarg-CS

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

@@ -39,7 +39,6 @@
 import io.cdap.plugin.common.ReferenceBatchSink;
 import io.cdap.plugin.common.ReferencePluginConfig;
 import io.cdap.plugin.common.batch.sink.SinkOutputFormatProvider;
-import io.cdap.plugin.common.util.ExceptionUtils;
 import io.cdap.plugin.db.ColumnType;
 import io.cdap.plugin.db.CommonSchemaReader;
 import io.cdap.plugin.db.ConnectionConfig;
@@ -59,7 +58,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.sql.Connection;
 import java.sql.Driver;
 import java.sql.PreparedStatement;
@@ -129,17 +127,8 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
     Class<? extends Driver> driverClass = DBUtils.getDriverClass(
       pipelineConfigurer, dbSinkConfig, ConnectionConfig.JDBC_PLUGIN_TYPE);
     if (driverClass != null && dbSinkConfig.canConnect()) {
-      try {
         validateSchema(collector, driverClass, dbSinkConfig.getTableName(), inputSchema,
           dbSinkConfig.getDBSchemaName());
-      } catch (SQLException e) {
-        String details = String.format("SQL error while validating the schema: Error: %s, SQLState: %s, ErrorCode: %s",
-                                       e.getMessage(), e.getSQLState(), e.getErrorCode());
-        collector.addFailure(details,
-          "Ensure the table exists and the connection string points to a valid database.")
-          .withConfigProperty(DBSinkConfig.TABLE_NAME)
-          .withStacktrace(e.getStackTrace());
-      }
     }
   }
   public void validateOperations(FailureCollector collector, T dbSinkConfig, @Nullable Schema inputSchema) {
@@ -229,9 +218,6 @@ public void prepareRun(BatchSinkContext context) {
       } else {
         outputSchema = inferSchema(driverClass);
       }
-    } catch (SQLException e) {
-      throw new RuntimeException(String.format("Unable to validate schema due to: %s.",
-        ExceptionUtils.getRootCauseMessage(e)), e);
     } finally {
       DBUtils.cleanup(driverClass);
     }
@@ -423,7 +409,7 @@ static List<ColumnType> getMatchedColumnTypeList(ResultSet resultSet, List<Strin
   }
 
   private void validateSchema(FailureCollector collector, Class<? extends Driver> jdbcDriverClass, String tableName,
-                              Schema inputSchema, String dbSchemaName) throws SQLException {
+                              Schema inputSchema, String dbSchemaName) {
     String connectionString = dbSinkConfig.getConnectionString();
     String fullyQualifiedTableName = dbSchemaName == null ? dbSinkConfig.getEscapedTableName()
             : dbSinkConfig.getEscapedDbSchemaName() + "." + dbSinkConfig.getEscapedTableName();
@@ -459,6 +445,13 @@ connectionString, connectionProperties, getExternalDocumentationLink())) {
              getExternalDocumentationLink())) {
         getFieldsValidator().validateFields(inputSchema, rs, collector);
       }
+    } catch (SQLException e) {
+      LOG.error("Exception while trying to validate schema of database table {} for connection {}.",
+              fullyQualifiedTableName, connectionString, e);
+      collector.addFailure(
+        String.format("Exception while trying to validate schema of database table '%s' for connection '%s' with %s",
+                fullyQualifiedTableName, connectionString, e.getMessage()),
+        null).withStacktrace(e.getStackTrace());
     }
   }