[BugFix] A race in LookPool may cause used Resources to be evicted

dora/core/common/src/main/java/alluxio/collections/LockPool.java

@@ -16,6 +16,7 @@
 import alluxio.resource.LockResource;
 import alluxio.resource.RWLockResource;
 import alluxio.resource.RefCountLockResource;
+import alluxio.util.AlluxioFaultInjector;
 import alluxio.util.ThreadFactoryUtils;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -165,8 +166,12 @@ private void awaitAndEvict() throws InterruptedException {
             candidate.mIsAccessed = false;
           } else {
             if (candidate.mRefCount.compareAndSet(0, Integer.MIN_VALUE)) {
-              mIterator.remove();
-              numToEvict--;
+              AlluxioFaultInjector.get().blockUtilAllocatedNewResource();
+              Resource tmpResource = mPool.compute(candidateMapEntry.getKey(),
+                  (k, v) -> v == candidate ? null : v);
+              if (tmpResource == null) {
+                numToEvict--;
+              }
             }
           }
         }

dora/core/common/src/main/java/alluxio/util/AlluxioFaultInjector.java

@@ -0,0 +1,38 @@
+/*
+ * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
+ * (the "License"). You may not use this work except in compliance with the License, which is
+ * available at www.apache.org/licenses/LICENSE-2.0
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied, as more fully set forth in the License.
+ *
+ * See the NOTICE file distributed with this work for information regarding copyright ownership.
+ */
+
+package alluxio.util;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * Used for injecting faults in Alluxio tests.
+ * Calls into this are a no-op in production code.
+ */
+@VisibleForTesting
+public class AlluxioFaultInjector {
+  private static AlluxioFaultInjector instance = new AlluxioFaultInjector();
+
+  public static AlluxioFaultInjector get() {
+    return instance;
+  }
+
+  public static void set(AlluxioFaultInjector injector) {
+    instance = injector;
+  }
+
+  /**
+   * Used as a hook to inject intercept when evicting unused resource from pool.
+   */
+  public void blockUtilAllocatedNewResource() {
+
+  }
+}

dora/core/common/src/test/java/alluxio/collections/LockPoolTest.java

@@ -16,12 +16,16 @@
 
 import alluxio.concurrent.LockMode;
 import alluxio.resource.LockResource;
+import alluxio.util.AlluxioFaultInjector;
 import alluxio.util.CommonUtils;
 
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 /**
@@ -141,4 +145,53 @@ public void referencedLockTest() {
     assertTrue(lock1.hasSameLock(mPool.get(50, LockMode.READ)));
     assertTrue(lock2.hasSameLock(mPool.get(100, LockMode.READ)));
   }
+
+  @Test(timeout = 10000)
+  public void evictorTest() throws InterruptedException {
+    List<LockResource> unClosedLocks = new ArrayList<>();
+    CountDownLatch mainThreadDownLatch = new CountDownLatch(1);
+    CountDownLatch evictorDownLatch = new CountDownLatch(1);
+    AlluxioFaultInjector.set(new AlluxioFaultInjector() {
+      public void blockUtilAllocatedNewResource() {
+        try {
+          mainThreadDownLatch.countDown();
+          evictorDownLatch.await();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    });
+
+    // Acquire 16 locks and just release the lock-3.
+    for (int i = 1; i <= 16; i++) {
+      LockResource lock = mPool.get(i, LockMode.READ);
+      if (i == 3) {
+        lock.close();
+      } else {
+        unClosedLocks.add(lock);
+      }
+    }
+    assertEquals(16, mPool.size());
+
+    // Acquire a new Lock to trigger evictor. And the Evictor will release the
+    // unused Resource-3, but it will be blocked by evictorDownLatch.
+    LockResource lock17 = mPool.get(17, LockMode.READ);
+    unClosedLocks.add(lock17);
+
+    // Acquire a resource for key 3 again, currently the Evictor is releasing
+    // the old Resource-3, but it is blocked right now.
+    mainThreadDownLatch.await();
+    LockResource lock3 = mPool.get(3, LockMode.READ);
+    unClosedLocks.add(lock3);
+
+    // Allow the evictor to continue.
+    evictorDownLatch.countDown();
+
+    // Sleep some millis to wait for the Evictor.
+    Thread.sleep(100);
+
+    // The Resource 3 should be still in the mPool.
+    assertEquals(17, mPool.size());
+    unClosedLocks.forEach(k -> k.close());
+  }
 }