Merge pull request #233 from data-integrations/exc-handling

CDAP-20472: Fix missing logging for exceptions

Author: sumitjnn

delta-plugins-common/src/main/java/io/cdap/delta/plugin/common/DriverCleanup.java

@@ -56,12 +56,13 @@ public static DriverCleanup ensureJDBCDriverIsAvailable(Class<? extends Driver>
       DriverManager.getDriver(url);
       return new DriverCleanup(null);
     } catch (SQLException e) {
+      LOG.warn("Error while getting JDBC Driver from URL {}", url, e);
       Driver driver = classz.newInstance();
       JDBCDriverShim shim = new JDBCDriverShim(driver);
       try {
         deregisterAllDrivers(classz);
       } catch (NoSuchFieldException | ClassNotFoundException e1) {
-        LOG.warn("Unable to deregister JDBC Driver class {}", classz);
+        LOG.warn("Unable to deregister JDBC Driver class {}", classz, e1);
       }
       DriverManager.registerDriver(shim);
       return new DriverCleanup(shim);

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlTableAssessor.java

@@ -28,6 +28,8 @@
 import io.cdap.delta.api.assessment.TableDetail;
 import io.cdap.delta.plugin.common.ColumnEvaluation;
 import io.cdap.delta.plugin.common.DriverCleanup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.sql.Connection;
@@ -44,6 +46,7 @@
  * MySQL table assessor.
  */
 public class MySqlTableAssessor implements TableAssessor<TableDetail> {
+  private static final Logger LOG = LoggerFactory.getLogger(MySqlTableAssessor.class);
   static final String COLUMN_LENGTH = "COLUMN_LENGTH";
   static final String SCALE = "SCALE";
 
@@ -197,6 +200,7 @@ private void checkReplicationPermissions(List<Problem> featureProblems, List<Str
         }
       }
     } catch (Exception e) {
+      LOG.error("Error in checking permissions", e);
       featureProblems.add(permissionUnknownProblem);
     }
   }

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerTableRegistry.java

@@ -52,7 +52,7 @@
  * Sql Server table registry
  */
 public class SqlServerTableRegistry implements TableRegistry {
-  private static final Logger LOGGER = LoggerFactory.getLogger(SqlServerTableRegistry.class);
+  private static final Logger LOG = LoggerFactory.getLogger(SqlServerTableRegistry.class);
   private final String jdbcUrl;
   private final SqlServerConfig config;
   private final DriverCleanup driverCleanup;
@@ -134,12 +134,12 @@ public TableDetail describeTable(String db, @Nullable String schema, String tabl
           }
         }
       } catch (Exception e) {
+        String msg = String.format("Unable to check if CDC feature for table '%s' in database '%s' was enabled or not",
+                                      schema == null ? table : schema + "." + table, db);
+        LOG.error(msg, e);
         missingFeatures.add(
-          new Problem("Unable To Check If CDC Was Enabled",
-                      String.format("Unable to check if CDC feature for table '%s' in database '%s' was enabled or not",
-                                    schema == null ? table : schema + "." + table, db),
-                      "Check database connectivity and table information",
-                      null));
+          new Problem("Unable To Check If CDC Was Enabled", msg,
+                      "Check database connectivity and table information", null));
       }
       return builder.setFeatures(missingFeatures).build();
     } catch (SQLException e) {
@@ -192,11 +192,11 @@ private Optional<TableDetail.Builder> getTableDetailBuilder(DatabaseMetaData dbM
       return Optional.empty();
     }
     List<String> primaryKey = new ArrayList<>();
-    LOGGER.debug("Query primary key for {}.{}.{}", db, schema, table);
+    LOG.debug("Query primary key for {}.{}.{}", db, schema, table);
     try (ResultSet keyResults = dbMeta.getPrimaryKeys(db, schema, table)) {
       while (keyResults.next()) {
         String pk = keyResults.getString("COLUMN_NAME");
-        LOGGER.debug("Found primary key for {}.{}.{} : {}", db, schema, table, pk);
+        LOG.debug("Found primary key for {}.{}.{} : {}", db, schema, table, pk);
         primaryKey.add(pk);
       }
     }