Merge pull request #139 from data-integrations/fix/sql-server-exception

CDAP-17570 Copy SQLServerErrorHandler class from debezium connector a…

Author: sagarkapare

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

@@ -28,6 +28,7 @@
 import io.debezium.connector.sqlserver.SourceInfo;
 import io.debezium.connector.sqlserver.SqlServerConnection;
 import io.debezium.connector.sqlserver.SqlServerConnector;
+import io.debezium.connector.sqlserver.SqlServerErrorHandler;
 import io.debezium.embedded.EmbeddedEngine;
 import io.debezium.jdbc.JdbcConfiguration;
 import io.debezium.jdbc.JdbcConnection;
@@ -85,6 +86,7 @@ public void start(Offset offset) {
                                                                      jdbcDriverClass.getName(),
                                                                      jdbcDriverClass.getClassLoader());
 
+    SqlServerErrorHandler.driverClassLoader = jdbcDriverClass.getClassLoader();
     // this is needed since sql server does not return the database information in the record
     String databaseName = config.getDatabase();
 

sqlserver-delta-plugins/src/main/java/io/debezium/connector/sqlserver/SqlServerErrorHandler.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.connector.sqlserver;
+
+import io.debezium.connector.base.ChangeEventQueue;
+import io.debezium.pipeline.ErrorHandler;
+
+/**
+ * Error handler for SQL Server.
+ *
+ * @author Chris Cranford
+ */
+public class SqlServerErrorHandler extends ErrorHandler {
+  // This class is copied from debezium. driverClassLoader variable is added to the
+  // original class. This class loader is the jdbc plugin class loader and is required
+  // to load the SQLServerException class from the user uploaded jdbc jar rather than looking
+  // into the debezium connector jar.
+  public static ClassLoader driverClassLoader;
+  public SqlServerErrorHandler(String logicalName, ChangeEventQueue<?> queue) {
+    super(SqlServerConnector.class, logicalName, queue);
+  }
+
+  @Override
+  protected boolean isRetriable(Throwable throwable) {
+    try {
+      Class<?> sqlServerExceptionClass = driverClassLoader.loadClass("com.microsoft.sqlserver.jdbc.SQLServerException");
+      return sqlServerExceptionClass.isInstance(throwable)
+        && (throwable.getMessage().contains("Connection timed out (Read failed)")
+        || throwable.getMessage().contains("The connection has been closed.")
+        || throwable.getMessage().contains("Connection reset")
+        || throwable.getMessage().contains("SHUTDOWN is in progress"));
+
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException("Failed to load the SQLServerException class.", e);
+    }
+  }
+}