Merge branch 'feat_1.8_checkConnection' into '1.8_test_3.10.x'

jdbc connection check



See merge request !247

Author: todd5167

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

@@ -170,6 +170,7 @@ private void checkConnectionOpen() {
             if (connection.isClosed()) {
                 LOG.info("db connection reconnect..");
                 establishConnection();
+                jdbcWriter.prepareStatement(connection);
             }
         } catch (SQLException e) {
             LOG.error("check connection open failed..", e);

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/AppendOnlyWriter.java

@@ -58,6 +58,11 @@ public AppendOnlyWriter(String insertSQL, int[] fieldTypes, DtRichOutputFormat m
     @Override
     public void open(Connection connection) throws SQLException {
         this.rows = new ArrayList();
+        prepareStatement(connection);
+    }
+
+    @Override
+    public void prepareStatement(Connection connection) throws SQLException {
         this.statement = connection.prepareStatement(insertSQL);
     }
 

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/JDBCWriter.java

@@ -33,10 +33,16 @@ public interface JDBCWriter extends Serializable {
 	int DIRTYDATA_PRINT_FREQUENTY = 1000;
 
 	/**
-	 * Open the writer by JDBC Connection. It can create Statement from Connection.
+	 * Open the writer by JDBC Connection.
 	 */
 	void open(Connection connection) throws SQLException;
 
+	/**
+	 *  Create Statement from Connection. Used where the connection is created
+	 * @throws SQLException
+	 */
+	void prepareStatement(Connection connection) throws SQLException;
+
 	/**
 	 * Add record to writer, the writer may cache the data.
 	 */

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/UpsertWriter.java

@@ -110,6 +110,11 @@ private UpsertWriter(int[] fieldTypes, int[] pkFields, int[] pkTypes, String del
     @Override
     public void open(Connection connection) throws SQLException {
         this.keyToRows = new HashMap<>();
+        prepareStatement(connection);
+    }
+
+    @Override
+    public void prepareStatement(Connection connection) throws SQLException {
         this.deleteStatement = connection.prepareStatement(deleteSQL);
     }
 
@@ -224,6 +229,11 @@ private UpsertWriterUsingUpsertStatement(
         @Override
         public void open(Connection connection) throws SQLException {
             super.open(connection);
+        }
+
+        @Override
+        public void prepareStatement(Connection connection) throws SQLException {
+            super.prepareStatement(connection);
             upsertStatement = connection.prepareStatement(upsertSQL);
         }
 
@@ -284,6 +294,11 @@ private UpsertWriterUsingInsertUpdateStatement(
         @Override
         public void open(Connection connection) throws SQLException {
             super.open(connection);
+        }
+
+        @Override
+        public void prepareStatement(Connection connection) throws SQLException {
+            super.prepareStatement(connection);
             existStatement = connection.prepareStatement(existSQL);
             insertStatement = connection.prepareStatement(insertSQL);
             updateStatement = connection.prepareStatement(updateSQL);