update concurrentMap version to v8

core/common/src/main/java/alluxio/collections/ConcurrentHashSet.java

@@ -11,25 +11,26 @@
 
 package alluxio.collections;
 
+import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
+
 import java.util.AbstractSet;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
- * A concurrent hash set. This is backed by a {@link ConcurrentHashMap}, and {@link Set} operations
- * are translated to {@link ConcurrentHashMap} operations.
+ * A concurrent hash set. This is backed by a {@link ConcurrentHashMapV8}, and {@link Set}
+ * operations are translated to {@link ConcurrentHashMapV8} operations.
  *
  * @param <T> the type of the set objects
  */
 @ThreadSafe
 public class ConcurrentHashSet<T> extends AbstractSet<T> {
   // COMPATIBILITY: This field needs to declared as Map (as opposed to ConcurrentHashMap). The
-  // reason is that the return type of ConcurrentHashMap#keySet() has changed from Set<K> to
+  // reason is that the return type of ConcurrentHashMapV8#keySet() has changed from Set<K> to
   // KeySetView<K,V> between Java 7 and Java 8 and this can result in a NoSuchMethod runtime
   // exception when using Java 7 to run byte code compiled with Java 8 (even if the compiler is
   // told to compile for Java 7).
@@ -52,7 +53,7 @@ public ConcurrentHashSet() {
    * @param concurrencyLevel the estimated number of concurrently updating threads
    */
   public ConcurrentHashSet(int initialCapacity, float loadFactor, int concurrencyLevel) {
-    mMap = new ConcurrentHashMap<>(initialCapacity, loadFactor, concurrencyLevel);
+    mMap = new ConcurrentHashMapV8<>(initialCapacity, loadFactor, concurrencyLevel);
   }
 
   @Override
@@ -77,9 +78,9 @@ public boolean add(T element) {
    * @return true if this set did not already contain the specified element
    */
   public boolean addIfAbsent(T element) {
-    // COMPATIBILITY: We need to cast mMap to ConcurrentHashMap to make sure the code can compile
+    // COMPATIBILITY: We need to cast mMap to ConcurrentHashMapV8 to make sure the code can compile
     // on Java 7 because the Map#putIfAbsent() method has only been introduced in Java 8.
-    return ((ConcurrentHashMap<T, Boolean>) mMap).putIfAbsent(element, Boolean.TRUE) == null;
+    return ((ConcurrentHashMapV8<T, Boolean>) mMap).putIfAbsent(element, Boolean.TRUE) == null;
   }
 
   @Override

core/common/src/main/java/alluxio/collections/NonUniqueFieldIndex.java

@@ -12,11 +12,11 @@
 package alluxio.collections;
 
 import com.google.common.collect.Iterables;
+import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
 
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -29,15 +29,15 @@
 @ThreadSafe
 public class NonUniqueFieldIndex<T> implements FieldIndex<T> {
   private final IndexDefinition<T> mIndexDefinition;
-  private final ConcurrentHashMap<Object, ConcurrentHashSet<T>> mIndexMap;
+  private final ConcurrentHashMapV8<Object, ConcurrentHashSet<T>> mIndexMap;
 
   /**
    * Constructs a new {@link NonUniqueFieldIndex} instance.
    *
    * @param indexDefinition definition of index
    */
   public NonUniqueFieldIndex(IndexDefinition<T> indexDefinition) {
-    mIndexMap = new ConcurrentHashMap<>(8, 0.95f, 8);
+    mIndexMap = new ConcurrentHashMapV8<>(8, 0.95f, 8);
     mIndexDefinition = indexDefinition;
   }
 

core/common/src/main/java/alluxio/collections/UniqueFieldIndex.java

@@ -11,10 +11,11 @@
 
 package alluxio.collections;
 
+import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
+
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -27,15 +28,15 @@
 @ThreadSafe
 public class UniqueFieldIndex<T> implements FieldIndex<T> {
   private final IndexDefinition<T> mIndexDefinition;
-  private final ConcurrentHashMap<Object, T> mIndexMap;
+  private final ConcurrentHashMapV8<Object, T> mIndexMap;
 
   /**
    * Constructs a new {@link UniqueFieldIndex} instance.
    *
    * @param indexDefinition definition of index
    */
   public UniqueFieldIndex(IndexDefinition<T> indexDefinition) {
-    mIndexMap = new ConcurrentHashMap<>(8, 0.95f, 8);
+    mIndexMap = new ConcurrentHashMapV8<>(8, 0.95f, 8);
     mIndexDefinition = indexDefinition;
   }
 

core/server/src/main/java/alluxio/master/block/BlockMaster.java

@@ -50,6 +50,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.protobuf.Message;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
 import org.apache.thrift.TProcessor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -64,7 +65,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -121,8 +121,8 @@ public Object getFieldValue(MasterWorkerInfo o) {
 
   // Block metadata management.
   /** Blocks on all workers, including active and lost blocks. This state must be journaled. */
-  private final ConcurrentHashMap<Long, MasterBlockInfo>
-      mBlocks = new ConcurrentHashMap<>(8192, 0.90f, 64);
+  private final ConcurrentHashMapV8<Long, MasterBlockInfo>
+      mBlocks = new ConcurrentHashMapV8<>(8192, 0.90f, 64);
   /** Keeps track of block which are no longer in Alluxio storage. */
   private final ConcurrentHashSet<Long> mLostBlocks = new ConcurrentHashSet<>(64, 0.90f, 64);
 

core/server/src/main/java/alluxio/worker/block/BlockLockManager.java

@@ -21,6 +21,7 @@
 import com.google.common.base.Objects;
 import com.google.common.base.Throwables;
 import com.google.common.collect.Sets;
+import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,7 +30,6 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -341,7 +341,7 @@ private void releaseBlockLockIfUnused(long blockId) {
   public void validate() {
     synchronized (mSharedMapsLock) {
       // Compute block lock reference counts based off of lock records
-      ConcurrentMap<Long, AtomicInteger> blockLockReferenceCounts = new ConcurrentHashMap<>();
+      ConcurrentMap<Long, AtomicInteger> blockLockReferenceCounts = new ConcurrentHashMapV8<>();
       for (LockRecord record : mLockIdToRecordMap.values()) {
         blockLockReferenceCounts.putIfAbsent(record.getBlockId(), new AtomicInteger(0));
         blockLockReferenceCounts.get(record.getBlockId()).incrementAndGet();

core/server/src/main/java/alluxio/worker/block/evictor/LRFUEvictor.java

@@ -24,6 +24,7 @@
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Iterators;
+import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -32,7 +33,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
 import javax.annotation.concurrent.NotThreadSafe;
@@ -51,9 +51,9 @@
 @NotThreadSafe
 public final class LRFUEvictor extends AbstractEvictor {
   // Map from block id to the last updated logic time count
-  private final Map<Long, Long> mBlockIdToLastUpdateTime = new ConcurrentHashMap<>();
+  private final Map<Long, Long> mBlockIdToLastUpdateTime = new ConcurrentHashMapV8<>();
   // Map from block id to the CRF value of the block
-  private final Map<Long, Double> mBlockIdToCRFValue = new ConcurrentHashMap<>();
+  private final Map<Long, Double> mBlockIdToCRFValue = new ConcurrentHashMapV8<>();
   // In the range of [0, 1]. Closer to 0, LRFU closer to LFU. Closer to 1, LRFU closer to LRU
   private final double mStepFactor;
   // In the range of [2, INF]