Switch back to Guava cache

Author: charlesconnell

hbase-common/pom.xml

@@ -110,10 +110,6 @@
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-crypto</artifactId>
     </dependency>
-    <dependency>
-      <groupId>com.github.ben-manes.caffeine</groupId>
-      <artifactId>caffeine</artifactId>
-    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/ByteBuffDecompressionCodec.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.io.compress;
 
+import java.io.IOException;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -27,7 +28,7 @@ public interface ByteBuffDecompressionCodec {
 
   ByteBuffDecompressor createByteBuffDecompressor();
 
-  Compression.HFileDecompressionContext
-    getDecompressionContextFromConfiguration(Configuration conf);
+  Compression.HFileDecompressionContext getDecompressionContextFromConfiguration(Configuration conf)
+    throws IOException;
 
 }

hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/CodecPool.java

@@ -17,8 +17,6 @@
  */
 package org.apache.hadoop.hbase.io.compress;
 
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.LoadingCache;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.util.Comparator;
 import java.util.NavigableSet;
@@ -37,6 +35,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.cache.CacheBuilder;
+import org.apache.hbase.thirdparty.com.google.common.cache.CacheLoader;
+import org.apache.hbase.thirdparty.com.google.common.cache.LoadingCache;
+
 /**
  * A global compressor/decompressor pool used to save and reuse (possibly native)
  * compression/decompression codecs. Copied from the class of the same name in hadoop-common and
@@ -56,7 +58,12 @@ public class CodecPool {
     NavigableSet<ByteBuffDecompressor>> BYTE_BUFF_DECOMPRESSOR_POOL = new ConcurrentHashMap<>();
 
   private static <T> LoadingCache<Class<T>, AtomicInteger> createCache() {
-    return Caffeine.newBuilder().build(key -> new AtomicInteger());
+    return CacheBuilder.newBuilder().build(new CacheLoader<>() {
+      @Override
+      public AtomicInteger load(Class<T> key) throws Exception {
+        return new AtomicInteger();
+      }
+    });
   }
 
   /**
@@ -108,26 +115,19 @@ private static <T> boolean payback(ConcurrentMap<Class<T>, NavigableSet<T>> pool
   /**
    * Copied from hadoop-common without significant modification.
    */
-  @SuppressWarnings("unchecked")
-  @edu.umd.cs.findbugs.annotations.SuppressWarnings(
-      value = "NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE",
-      justification = "LoadingCache will compute value if absent")
   private static <T> int getLeaseCount(LoadingCache<Class<T>, AtomicInteger> usageCounts,
     Class<? extends T> codecClass) {
-    return usageCounts.get((Class<T>) codecClass).get();
+    return usageCounts.getUnchecked((Class<T>) codecClass).get();
   }
 
   /**
    * Copied from hadoop-common without significant modification.
    */
-  @edu.umd.cs.findbugs.annotations.SuppressWarnings(
-      value = "NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE",
-      justification = "LoadingCache will compute value if absent")
   private static <T> void updateLeaseCount(LoadingCache<Class<T>, AtomicInteger> usageCounts,
     T codec, int delta) {
     if (codec != null && usageCounts != null) {
       Class<T> codecClass = ReflectionUtils.getClass(codec);
-      usageCounts.get(codecClass).addAndGet(delta);
+      usageCounts.getUnchecked(codecClass).addAndGet(delta);
     }
   }
 

hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/Compression.java

@@ -561,7 +561,7 @@ public void returnByteBuffDecompressor(ByteBuffDecompressor decompressor) {
      */
     @Nullable
     public HFileDecompressionContext
-      getHFileDecompressionContextForConfiguration(Configuration conf) {
+      getHFileDecompressionContextForConfiguration(Configuration conf) throws IOException {
       if (supportsByteBuffDecompression()) {
         return ((ByteBuffDecompressionCodec) getCodec(conf))
           .getDecompressionContextFromConfiguration(conf);

hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/DictionaryCache.java

@@ -17,12 +17,11 @@
  */
 package org.apache.hadoop.hbase.io.compress;
 
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
 import java.io.ByteArrayOutputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
@@ -32,6 +31,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.cache.CacheBuilder;
+import org.apache.hbase.thirdparty.com.google.common.cache.CacheLoader;
+import org.apache.hbase.thirdparty.com.google.common.cache.LoadingCache;
+
 /**
  * A utility class for managing compressor/decompressor dictionary loading and caching of load
  * results. Useful for any codec that can support changing dictionaries at runtime, such as
@@ -45,8 +48,7 @@ public final class DictionaryCache {
   public static final String RESOURCE_SCHEME = "resource://";
 
   private static final Logger LOG = LoggerFactory.getLogger(DictionaryCache.class);
-  private static final Cache<String, byte[]> BYTE_ARRAY_CACHE =
-    Caffeine.newBuilder().maximumSize(100).expireAfterAccess(10, TimeUnit.MINUTES).build();
+  private static volatile LoadingCache<String, byte[]> CACHE;
 
   private DictionaryCache() {
   }
@@ -62,22 +64,35 @@ public static byte[] getDictionary(final Configuration conf, final String path)
       return null;
     }
 
-    // Get or load the dictionary for the given path
-    return BYTE_ARRAY_CACHE.get(path, s -> {
-      final int maxSize = conf.getInt(DICTIONARY_MAX_SIZE_KEY, DEFAULT_DICTIONARY_MAX_SIZE);
-      byte[] bytes;
-      try {
-        if (path.startsWith(RESOURCE_SCHEME)) {
-          bytes = loadFromResource(conf, path, maxSize);
-        } else {
-          bytes = loadFromHadoopFs(conf, path, maxSize);
+    // Create the dictionary loading cache if we haven't already
+    if (CACHE == null) {
+      synchronized (DictionaryCache.class) {
+        if (CACHE == null) {
+          final int maxSize = conf.getInt(DICTIONARY_MAX_SIZE_KEY, DEFAULT_DICTIONARY_MAX_SIZE);
+          CACHE = CacheBuilder.newBuilder().maximumSize(100).expireAfterAccess(10, TimeUnit.MINUTES)
+            .build(new CacheLoader<String, byte[]>() {
+              @Override
+              public byte[] load(String s) throws Exception {
+                byte[] bytes;
+                if (path.startsWith(RESOURCE_SCHEME)) {
+                  bytes = loadFromResource(conf, path, maxSize);
+                } else {
+                  bytes = loadFromHadoopFs(conf, path, maxSize);
+                }
+                LOG.info("Loaded dictionary from {} (size {})", s, bytes.length);
+                return bytes;
+              }
+            });
         }
-      } catch (IOException e) {
-        throw new RuntimeException(e);
       }
-      LOG.info("Loaded dictionary from {} (size {})", s, bytes.length);
-      return bytes;
-    });
+    }
+
+    // Get or load the dictionary for the given path
+    try {
+      return CACHE.get(path);
+    } catch (ExecutionException e) {
+      throw new RuntimeException("Unable to load dictionary at " + path, e);
+    }
   }
 
   // Visible for testing
@@ -86,7 +101,7 @@ public static byte[] loadFromResource(final Configuration conf, final String s,
     if (!s.startsWith(RESOURCE_SCHEME)) {
       throw new IOException("Path does not start with " + RESOURCE_SCHEME);
     }
-    final String path = s.substring(RESOURCE_SCHEME.length());
+    final String path = s.substring(RESOURCE_SCHEME.length(), s.length());
     LOG.info("Loading resource {}", path);
     final InputStream in = DictionaryCache.class.getClassLoader().getResourceAsStream(path);
     if (in == null) {
@@ -140,7 +155,10 @@ private static byte[] loadFromHadoopFs(final Configuration conf, final String s,
 
   // Visible for testing
   public static boolean contains(String dictionaryPath) {
-    return BYTE_ARRAY_CACHE.asMap().containsKey(dictionaryPath);
+    if (CACHE != null) {
+      return CACHE.asMap().containsKey(dictionaryPath);
+    }
+    return false;
   }
 
 }

hbase-compression/hbase-compression-zstd/src/main/java/org/apache/hadoop/hbase/io/compress/zstd/ZstdCodec.java

@@ -17,8 +17,6 @@
  */
 package org.apache.hadoop.hbase.io.compress.zstd;
 
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
 import com.github.luben.zstd.Zstd;
 import com.github.luben.zstd.ZstdDictDecompress;
 import edu.umd.cs.findbugs.annotations.Nullable;
@@ -27,6 +25,7 @@
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
@@ -45,6 +44,9 @@
 import org.apache.hadoop.io.compress.Decompressor;
 import org.apache.yetus.audience.InterfaceAudience;
 
+import org.apache.hbase.thirdparty.com.google.common.cache.Cache;
+import org.apache.hbase.thirdparty.com.google.common.cache.CacheBuilder;
+
 /**
  * Hadoop ZStandard codec implemented with zstd-jni.
  * <p>
@@ -59,7 +61,7 @@ public class ZstdCodec implements Configurable, CompressionCodec, ByteBuffDecomp
   public static final String ZSTD_DICTIONARY_KEY = "hbase.io.compress.zstd.dictionary";
 
   private static final Cache<String, Pair<ZstdDictDecompress, Integer>> DECOMPRESS_DICT_CACHE =
-    Caffeine.newBuilder().maximumSize(100).expireAfterAccess(10, TimeUnit.MINUTES).build();
+    CacheBuilder.newBuilder().maximumSize(100).expireAfterAccess(10, TimeUnit.MINUTES).build();
 
   private Configuration conf;
   private int bufferSize;
@@ -137,7 +139,7 @@ public Class<? extends ByteBuffDecompressor> getByteBuffDecompressorType() {
 
   @Override
   public Compression.HFileDecompressionContext
-    getDecompressionContextFromConfiguration(Configuration conf) {
+    getDecompressionContextFromConfiguration(Configuration conf) throws IOException {
     return ZstdHFileDecompressionContext.fromConfiguration(conf);
   }
 
@@ -177,11 +179,15 @@ static Pair<ZstdDictDecompress, Integer> getDecompressDictionary(final Configura
       return null;
     }
 
-    return DECOMPRESS_DICT_CACHE.get(path, (s) -> {
-      byte[] dictBytes = DictionaryCache.getDictionary(conf, path);
-      int dictId = getDictionaryId(dictBytes);
-      return new Pair<>(new ZstdDictDecompress(dictBytes), dictId);
-    });
+    try {
+      return DECOMPRESS_DICT_CACHE.get(path, () -> {
+        byte[] dictBytes = DictionaryCache.getDictionary(conf, path);
+        int dictId = getDictionaryId(dictBytes);
+        return new Pair<>(new ZstdDictDecompress(dictBytes), dictId);
+      });
+    } catch (ExecutionException e) {
+      throw new RuntimeException("Unable to load ZSTD dictionary", e);
+    }
   }
 
   // Zstandard dictionaries begin with a 32-bit magic number, 0xEC30A437 in little-endian

hbase-compression/hbase-compression-zstd/src/main/java/org/apache/hadoop/hbase/io/compress/zstd/ZstdHFileDecompressionContext.java

@@ -63,7 +63,8 @@ public boolean isAllowByteBuffDecompression() {
     return allowByteBuffDecompression;
   }
 
-  public static ZstdHFileDecompressionContext fromConfiguration(Configuration conf) {
+  public static ZstdHFileDecompressionContext fromConfiguration(Configuration conf)
+    throws IOException {
     boolean allowByteBuffDecompression =
       conf.getBoolean("hbase.io.compress.zstd.allowByteBuffDecompression", true);
     Pair<ZstdDictDecompress, Integer> dictAndId = ZstdCodec.getDecompressDictionary(conf);