db reconnection

Author: todd5167

pom.xml

@@ -10,19 +10,19 @@
     <url>http://maven.apache.org</url>
     <modules>
         <module>core</module>
-        <module>kafka09</module>
-        <module>kafka10</module>
+        <!--<module>kafka09</module>-->
+        <!--<module>kafka10</module>-->
         <module>kafka11</module>
         <module>mysql</module>
-        <module>hbase</module>
-        <module>elasticsearch5</module>
-        <module>mongo</module>
-        <module>redis5</module>
+        <!--<module>hbase</module>-->
+        <!--<module>elasticsearch5</module>-->
+        <!--<module>mongo</module>-->
+        <!--<module>redis5</module>-->
         <module>launcher</module>
         <module>rdb</module>
-        <module>sqlserver</module>
-        <module>oracle</module>
-        <module>cassandra</module>
+        <!--<module>sqlserver</module>-->
+        <!--<module>oracle</module>-->
+        <!--<module>cassandra</module>-->
     </modules>
 
 

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

@@ -85,7 +85,7 @@ public void configure(Configuration parameters) {
     @Override
     public void open(int taskNumber, int numTasks) throws IOException {
         try {
-            establishConnection();
+            dbConn = establishConnection();
             initMetric();
             if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
                 if (isReplaceInsertQuery()) {
@@ -104,13 +104,15 @@ public void open(int taskNumber, int numTasks) throws IOException {
     }
 
 
-    private void establishConnection() throws SQLException, ClassNotFoundException {
+    private Connection establishConnection() throws SQLException, ClassNotFoundException {
+        Connection connection ;
         Class.forName(drivername);
         if (username == null) {
-            dbConn = DriverManager.getConnection(dbURL);
+            connection = DriverManager.getConnection(dbURL);
         } else {
-            dbConn = DriverManager.getConnection(dbURL, username, password);
+            connection = DriverManager.getConnection(dbURL, username, password);
         }
+        return connection;
     }
 
     /**
@@ -151,7 +153,7 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
 
 
     private void insertWrite(Row row) throws SQLException {
-
+        checkConnectionOpen(dbConn);
         updatePreparedStmt(row, upload);
         upload.addBatch();
         batchCount++;
@@ -161,6 +163,20 @@ private void insertWrite(Row row) throws SQLException {
         }
     }
 
+    private void checkConnectionOpen(Connection dbConn) {
+        try {
+            if (dbConn.isClosed()) {
+                LOG.info("db connection reconnect..");
+                dbConn= establishConnection();
+                upload = dbConn.prepareStatement(insertQuery);
+            }
+        } catch (SQLException e) {
+            LOG.error("check connection open failed..", e);
+        } catch (ClassNotFoundException e) {
+            LOG.error("load jdbc class error when reconnect db..", e);
+        }
+    }
+
     private void updatePreparedStmt(Row row, PreparedStatement pstmt) throws SQLException {
         if (typesArray == null) {
             // no types provided