optimize: undo log dirty throw BranchRollbackFailed_Unretriable

rm-datasource/src/main/java/io/seata/rm/datasource/undo/AbstractUndoExecutor.java

@@ -273,7 +273,7 @@ protected boolean dataValidationAndGoOn(Connection conn) throws SQLException {
                             "oldRows:[" + JSON.toJSONString(afterRecords.getRows()) + "]," +
                             "newRows:[" + JSON.toJSONString(currentRecords.getRows()) + "].");
                 }
-                throw new SQLException("Has dirty records when undo.");
+                throw new SQLUndoDirtyException("Has dirty records when undo.");
             }
         }
         return true;

rm-datasource/src/main/java/io/seata/rm/datasource/undo/AbstractUndoLogManager.java

@@ -49,6 +49,7 @@
 import static io.seata.common.DefaultValues.DEFAULT_CLIENT_UNDO_COMPRESS_TYPE;
 import static io.seata.common.DefaultValues.DEFAULT_CLIENT_UNDO_COMPRESS_THRESHOLD;
 import static io.seata.core.exception.TransactionExceptionCode.BranchRollbackFailed_Retriable;
+import static io.seata.core.exception.TransactionExceptionCode.BranchRollbackFailed_Unretriable;
 
 /**
  * @author jsbxyyx
@@ -358,9 +359,17 @@ public void undo(DataSourceProxy dataSourceProxy, String xid, long branchId) thr
                         LOGGER.warn("Failed to close JDBC resource while undo ... ", rollbackEx);
                     }
                 }
-                throw new BranchTransactionException(BranchRollbackFailed_Retriable, String
-                    .format("Branch session rollback failed and try again later xid = %s branchId = %s %s", xid,
-                        branchId, e.getMessage()), e);
+                if (e instanceof SQLUndoDirtyException) {
+                    throw new BranchTransactionException(BranchRollbackFailed_Unretriable,
+                        String.format(
+                            "Branch session rollback failed because of dirty undo log. xid = %s branchId = %s", xid,
+                            branchId),
+                        e);
+                }
+                throw new BranchTransactionException(BranchRollbackFailed_Retriable,
+                    String.format("Branch session rollback failed and try again later xid = %s branchId = %s %s", xid,
+                        branchId, e.getMessage()),
+                    e);
 
             } finally {
                 try {

rm-datasource/src/main/java/io/seata/rm/datasource/undo/SQLUndoDirtyException.java

@@ -0,0 +1,31 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed 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 io.seata.rm.datasource.undo;
+
+import java.io.Serializable;
+import java.sql.SQLException;
+
+/**
+ * @author zouwei
+ */
+class SQLUndoDirtyException extends SQLException implements Serializable {
+
+    private static final long serialVersionUID = -5168905669539637570L;
+
+    SQLUndoDirtyException(String reason) {
+        super(reason);
+    }
+}