deal rdb sink error

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/enums/EConnectionErrorCode.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.enums;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.exception.ExceptionUtils;
+
+import java.util.Arrays;
+
+/**
+ * Date: 2020/4/2
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public enum EConnectionErrorCode {
+    ERROR_NOT_MATCH(0, "错误信息未匹配", new String[]{}),
+    CONN_DB_INVALID(1, "数据库连接失效，请重新打开", new String[]{"the last packet successfully received from the server was"}),
+    CONN_DB_FAILED(2, "数据库连接失败，请检查用户名或密码是否正确", new String[]{"Access denied for user"}),
+    DB_TABLE_NOT_EXIST(3, "操作的表名不存在", new String[]{"doesn't exist"});
+
+    private int code;
+    private String description;
+    private String[] baseErrorInfo;
+
+    EConnectionErrorCode(int code, String description, String[] baseErrorInfo) {
+        this.code = code;
+        this.description = description;
+        this.baseErrorInfo = baseErrorInfo;
+    }
+
+
+    public static EConnectionErrorCode resolveErrorCodeFromException(Throwable e) {
+        final String stackErrorMsg = ExceptionUtils.getFullStackTrace(e);
+        return Arrays.stream(values())
+                .filter(errorCode -> matchKnowError(errorCode, stackErrorMsg))
+                .findAny()
+                .orElse(ERROR_NOT_MATCH);
+    }
+
+    public static boolean matchKnowError(EConnectionErrorCode errorCode, String errorMsg) {
+        return Arrays.stream(errorCode.baseErrorInfo)
+                .filter(baseInfo -> StringUtils.containsIgnoreCase(errorMsg, baseInfo))
+                .findAny()
+                .isPresent();
+    }
+}

core/src/test/java/com/dtstack/flink/sql/side/EConnectionErrorCodeTest.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.side;
+
+import com.dtstack.flink.sql.enums.EConnectionErrorCode;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Date: 2020/4/2
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class EConnectionErrorCodeTest {
+
+    @Test
+    public void testResolveErrorCodeFromException(){
+        EConnectionErrorCode errorCode =
+                EConnectionErrorCode.resolveErrorCodeFromException(new Exception("The last packet successfully received from the server was 179 milliseconds"));
+
+        Assert.assertEquals(errorCode, EConnectionErrorCode.CONN_DB_INVALID);
+    }
+}

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

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.sink.rdb.format;
 
+import com.dtstack.flink.sql.enums.EConnectionErrorCode;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
 import com.dtstack.flink.sql.util.JDBCUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -181,7 +182,7 @@ public void writeRecord(Tuple2 tuple2)  {
     }
 
 
-    private void insertWrite(Row row) {
+    private void insertWrite(Row row)  {
         try {
             if (batchNum == 1) {
                 writeSingleRecord(row);
@@ -201,8 +202,6 @@ private void insertWrite(Row row) {
             } else {
                 outDirtyRecords.inc(batchNum == 1 ? batchNum : rows.size());
             }
-
-
         }
 
     }
@@ -213,7 +212,7 @@ private void writeSingleRecord(Row row) {
             upload.executeUpdate();
             dbConn.commit();
         } catch (SQLException e) {
-
+            dealSQLException(e);
             if (outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0 || LOG.isDebugEnabled()) {
                 LOG.error("record insert failed,dirty record num:{}, current row:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
@@ -315,46 +314,59 @@ private void updatePreparedStmt(Row row, PreparedStatement pstmt) throws SQLExce
 
     private synchronized void submitExecuteBatch() {
         try {
-            checkConnectionOpen();
+            regularlyCheckConnection();
             this.upload.executeBatch();
             dbConn.commit();
         } catch (SQLException e) {
+            LOG.warn("submitExecuteBatch error {}", e);
             try {
                 dbConn.rollback();
             } catch (SQLException e1) {
-                LOG.error("rollback  data error !", e);
+                dealSQLException(e1);
             }
-
             rows.forEach(this::writeSingleRecord);
         } finally {
             rows.clear();
         }
     }
 
-    private void checkConnectionOpen() {
-        LOG.info("test db connection Valid check !");
+    private void dealSQLException(Exception e) {
+        EConnectionErrorCode errorCode = EConnectionErrorCode.resolveErrorCodeFromException(e);
+        switch (errorCode) {
+            case CONN_DB_INVALID:
+                reconnection();
+                break;
+            case CONN_DB_FAILED:
+            case DB_TABLE_NOT_EXIST:
+                throw new RuntimeException(e);
+            default:
+        }
+    }
+
+
+    private void regularlyCheckConnection() throws SQLException {
         checkTimes++;
         if (checkTimes % CONNECTION_CHECK_FREQUENCY != 0) {
             return;
         }
         LOG.warn("db connection Valid check !");
-        try {
-            if (dbConn.isClosed() || !dbConn.isValid(100)) {
-                LOG.info("db connection reconnect..");
-                dbConn = establishConnection();
-                upload = dbConn.prepareStatement(insertQuery);
-                this.dbConn = dbConn;
-            }
-        } catch (SQLException e) {
-            LOG.error("check connection open failed..", e);
-        } catch (ClassNotFoundException e) {
-            LOG.error("load jdbc class error when reconnect db..", e);
-        } catch (IOException e) {
-            LOG.error("kerberos authentication failed..", e);
+        if (dbConn.isClosed() || !dbConn.isValid(100)) {
+            reconnection();
         }
         checkTimes = 0;
     }
 
+    public void reconnection() throws RuntimeException {
+        try {
+            LOG.info("db connection reconnect..");
+            dbConn = establishConnection();
+            upload = dbConn.prepareStatement(insertQuery);
+            this.dbConn = dbConn;
+        } catch (Exception e) {
+            throw new RuntimeException("connection open failed..", e);
+        }
+    }
+
     /**
      * Executes prepared statement and closes all resources of this instance.
      *