bugfix: filter repeated lock (apache#1839)

dockerDo · Oct 31, 2019 · 5730057 · 5730057
1 parent 9dab05d
commit 5730057
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 4 deletions.
diff --git a/.../src/main/resources/protobuf/io/seata/protocol/transcation/transactionExceptionCode.proto b/.../src/main/resources/protobuf/io/seata/protocol/transcation/transactionExceptionCode.proto
@@ -109,5 +109,9 @@ enum TransactionExceptionCodeProto {
      */
     FailedWriteSession = 16;
 
+    /**
+     * FailedStore
+     */
+    FailedStore = 17;
 
 }
diff --git a/common/src/main/java/io/seata/common/util/LambdaUtils.java b/common/src/main/java/io/seata/common/util/LambdaUtils.java
@@ -0,0 +1,21 @@
+package io.seata.common.util;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+/**
+ * The type Lambda util.
+ *
+ * @author zjinlei
+ * @date 2019 /10/29
+ */
+public class LambdaUtils {
+
+    public static <T> Predicate<T> distinctByKey(Function<? super T, Object> keyExtractor) {
+        Map<Object, Boolean> seen = new ConcurrentHashMap<>();
+        return object -> seen.putIfAbsent(keyExtractor.apply(object), Boolean.TRUE) == null;
+    }
+
+}
diff --git a/core/src/main/java/io/seata/core/store/db/LockStoreDataBaseDAO.java b/core/src/main/java/io/seata/core/store/db/LockStoreDataBaseDAO.java
@@ -15,6 +15,7 @@
  */
 package io.seata.core.store.db;
 
+import javax.sql.DataSource;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
@@ -25,13 +26,12 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import javax.sql.DataSource;
-
 import io.seata.common.exception.DataAccessException;
 import io.seata.common.exception.StoreException;
 import io.seata.common.executor.Initialize;
 import io.seata.common.loader.LoadLevel;
 import io.seata.common.util.CollectionUtils;
+import io.seata.common.util.LambdaUtils;
 import io.seata.common.util.StringUtils;
 import io.seata.config.Configuration;
 import io.seata.config.ConfigurationFactory;
@@ -104,9 +104,11 @@ public boolean acquireLock(List<LockDO> lockDOs) {
         Connection conn = null;
         PreparedStatement ps = null;
         ResultSet rs = null;
-        List<LockDO> unrepeatedLockDOs = null;
         Set<String> dbExistedRowKeys = new HashSet<>();
         boolean originalAutoCommit = true;
+        if (lockDOs.size() > 1) {
+            lockDOs = lockDOs.stream().filter(LambdaUtils.distinctByKey(LockDO::getRowKey)).collect(Collectors.toList());
+        }
         try {
             conn = logStoreDataSource.getConnection();
             if (originalAutoCommit = conn.getAutoCommit()) {
@@ -149,6 +151,7 @@ public boolean acquireLock(List<LockDO> lockDOs) {
                 conn.rollback();
                 return false;
             }
+            List<LockDO> unrepeatedLockDOs = null;
             if (CollectionUtils.isNotEmpty(dbExistedRowKeys)) {
                 unrepeatedLockDOs = lockDOs.stream().filter(lockDO -> !dbExistedRowKeys.contains(lockDO.getRowKey()))
                     .collect(Collectors.toList());
@@ -159,7 +162,6 @@ public boolean acquireLock(List<LockDO> lockDOs) {
                 conn.rollback();
                 return true;
             }
-
             //lock
             for (LockDO lockDO : unrepeatedLockDOs) {
                 if (!doAcquireLock(conn, lockDO)) {