mysql log

Author: todd5167

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

@@ -98,6 +98,15 @@ public void open(int taskNumber, int numTasks) throws IOException {
             establishConnection();
             initMetric();
 
+            if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
+                if (isReplaceInsertQuery()) {
+                    insertQuery = dbSink.buildUpdateSql(tableName, Arrays.asList(dbSink.getFieldNames()), realIndexes, fullField);
+                }
+                upload = dbConn.prepareStatement(insertQuery);
+            } else {
+                throw new SQLException("Table " + tableName + " doesn't exist");
+            }
+
             if (batchWaitInterval > 0) {
                 LOG.info("open batch wait interval scheduled, interval is {} ms", batchWaitInterval);
 
@@ -108,15 +117,6 @@ public void open(int taskNumber, int numTasks) throws IOException {
 
             }
 
-            if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
-                if (isReplaceInsertQuery()) {
-                    insertQuery = dbSink.buildUpdateSql(tableName, Arrays.asList(dbSink.getFieldNames()), realIndexes, fullField);
-                }
-                upload = dbConn.prepareStatement(insertQuery);
-            } else {
-                throw new SQLException("Table " + tableName + " doesn't exist");
-            }
-
         } catch (SQLException sqe) {
             LOG.error("", sqe);
             throw new IllegalArgumentException("open() failed.", sqe);