Add catch block for inferSchema and add Error Code and SQLState to validateSchema

Author: sgarg-CS

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

@@ -318,6 +318,10 @@ private Schema inferSchema(Class<? extends Driver> driverClass) {
                String.format("SELECT * FROM %s WHERE 1 = 0", fullyQualifiedTableName), getErrorDetailsProvider())) {
           inferredFields.addAll(getSchemaReader().getSchemaFields(rs));
         }
+      } catch (SQLException e) {
+        LOG.error("Exception while trying to infer schema of database table {} for connection {}. [SQLState={}, " +
+          "ErrorCode={}]", fullyQualifiedTableName, dbSinkConfig.getConnectionString(), e.getSQLState(),
+            e.getErrorCode(), e);
       }
     } catch (IllegalAccessException | InstantiationException | SQLException e) {
       throw new InvalidStageException("JDBC Driver unavailable: " + dbSinkConfig.getJdbcPluginName(), e);
@@ -450,12 +454,12 @@ connectionString, connectionProperties, getErrorDetailsProvider())) {
         getFieldsValidator().validateFields(inputSchema, rs, collector);
       }
     } catch (SQLException e) {
-      LOG.error("Exception while trying to validate schema of database table {} for connection {}.",
-              fullyQualifiedTableName, connectionString, e);
+      LOG.error("Exception while trying to validate schema of database table {} for connection {}. [SQLState={}," +
+        " ErrorCode={}]", fullyQualifiedTableName, connectionString, e.getSQLState(), e.getErrorCode(), 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());
+        String.format("Exception while trying to validate schema of database table '%s' for connection '%s' with %s. "
+          + "[SQLState='%s', ErrorCode='%s']", fullyQualifiedTableName, connectionString, e.getMessage(),
+            e.getSQLState(), e.getErrorCode()), null).withStacktrace(e.getStackTrace());
     }
   }