Replace Guava with Caffeine and additional api to ComposableFuture/s

.travis.yml

@@ -11,4 +11,4 @@ before_script:
   - ./consul --version
 
 script:
-  - ../consul agent -dev -advertise=127.0.0.1 &
+  - ../consul agent -dev -advertise=127.0.0.1 &

ob1k-cache/pom.xml

@@ -55,6 +55,11 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>com.github.ben-manes.caffeine</groupId>
+      <artifactId>caffeine</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-log4j12</artifactId>
@@ -82,5 +87,11 @@
       <groupId>net.jpountz.lz4</groupId>
       <artifactId>lz4</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>org.jetbrains.kotlin</groupId>
+      <artifactId>kotlin-stdlib-jdk8</artifactId>
+    </dependency>
+
   </dependencies>
 </project>

ob1k-cache/src/main/java/com/outbrain/ob1k/cache/CacheConfiguration.kt

@@ -0,0 +1,61 @@
+package com.outbrain.ob1k.cache
+
+
+import com.outbrain.swinfra.metrics.api.MetricFactory
+import java.util.concurrent.TimeUnit
+
+data class CacheConfiguration<K, V>(val cacheName: String,
+                                    var ttl: Int = 360_000,
+                                    var ttlTimeUnit: TimeUnit = TimeUnit.MILLISECONDS,
+                                    var loader: CacheLoader<K, V>? = null,
+                                    var failOnMissingEntries: Boolean = false,
+                                    var maxSize: Int = 1_000_000,
+                                    var loadTimeout: Long = 500,
+                                    var loadTimeUnit: TimeUnit = TimeUnit.MILLISECONDS,
+                                    var metricFactory: MetricFactory? = null) {
+
+  constructor(cacheName1: String) : this(cacheName = cacheName1)
+
+  @JvmOverloads
+  fun withTtl(ttl: Int, ttlTimeUnit: TimeUnit = TimeUnit.MILLISECONDS): CacheConfiguration<K, V> {
+    this.ttl = ttl
+    this.ttlTimeUnit = ttlTimeUnit
+    return this
+  }
+
+  fun withLoader(loader: CacheLoader<K, V>): CacheConfiguration<K, V> {
+    this.loader = loader
+    return this
+  }
+
+  @JvmOverloads
+  fun failOnMissingEntries(fail: Boolean = true): CacheConfiguration<K, V> {
+    this.failOnMissingEntries = fail
+    return this
+  }
+
+  fun withMaxSize(maxSize: Int): CacheConfiguration<K, V> {
+    this.maxSize = maxSize
+    return this
+  }
+
+  @JvmOverloads
+  fun withLoadTimeout(loadTimeout: Long, loadTimeUnit: TimeUnit = TimeUnit.MILLISECONDS): CacheConfiguration<K, V> {
+    this.loadTimeout = loadTimeout
+    this.loadTimeUnit = loadTimeUnit
+    return this
+  }
+
+  /**
+   * for testing purposes
+   */
+  fun withMetricFactory(metricFactory: MetricFactory): CacheConfiguration<K, V> {
+    this.metricFactory = metricFactory
+    return this
+  }
+
+  fun buildLocalAsyncCache(): LocalAsyncCache<K, V> = LocalAsyncCache(this)
+
+  fun buildLoadingCacheDelegate(cache: TypedCache<K, V>) = LoadingCacheDelegate(cache, this)
+
+}

ob1k-cache/src/main/java/com/outbrain/ob1k/cache/GuavaCacheGaugesFactory.java → ob1k-cache/src/main/java/com/outbrain/ob1k/cache/CaffeineCacheGaugesFactory.java

@@ -1,6 +1,6 @@
 package com.outbrain.ob1k.cache;
 
-import com.google.common.cache.Cache;
+import com.github.benmanes.caffeine.cache.Cache;
 import com.outbrain.swinfra.metrics.api.MetricFactory;
 
 
@@ -9,14 +9,14 @@
  * creates a set of gauges that monitors Guava cache.
  *
  */
-public class GuavaCacheGaugesFactory {
+public class CaffeineCacheGaugesFactory {
   public static void createGauges(final MetricFactory metricFactory, final Cache cache, final String cacheName) {
     if (metricFactory != null) {
       metricFactory.registerGauge(cacheName, "averageLoadPenalty", () -> cache.stats().averageLoadPenalty());
 
       metricFactory.registerGauge(cacheName, "hitRate", () -> cache.stats().hitRate());
 
-      metricFactory.registerGauge(cacheName, "loadExceptionRate", () -> cache.stats().loadExceptionRate());
+      metricFactory.registerGauge(cacheName, "loadFailureRate", () -> cache.stats().loadFailureRate());
 
       metricFactory.registerGauge(cacheName, "missRate", () -> cache.stats().missRate());
 
@@ -26,7 +26,7 @@ public static void createGauges(final MetricFactory metricFactory, final Cache c
 
       metricFactory.registerGauge(cacheName, "loadCount", () -> cache.stats().loadCount());
 
-      metricFactory.registerGauge(cacheName, "loadExceptionCount", () -> cache.stats().loadExceptionCount());
+      metricFactory.registerGauge(cacheName, "loadFailureCount", () -> cache.stats().loadFailureCount());
 
       metricFactory.registerGauge(cacheName, "loadSuccessCount", () -> cache.stats().loadSuccessCount());
 
@@ -36,7 +36,7 @@ public static void createGauges(final MetricFactory metricFactory, final Cache c
 
       metricFactory.registerGauge(cacheName, "totalLoadTime", () -> cache.stats().totalLoadTime());
 
-      metricFactory.registerGauge(cacheName, "size", cache::size);
+      metricFactory.registerGauge(cacheName, "size", cache::estimatedSize);
 
     }
 

ob1k-cache/src/main/java/com/outbrain/ob1k/cache/LoadingCacheDelegate.java

@@ -64,6 +64,7 @@ public LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V
     this(cache, loader, cacheName, metricFactory, duration, timeUnit, false);
   }
 
+  @Deprecated
   public LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V> loader, final String cacheName,
                               final MetricFactory metricFactory, final long duration, final TimeUnit timeUnit,
                               final boolean failOnError) {
@@ -96,6 +97,38 @@ public LoadingCacheDelegate(final TypedCache<K, V> cache, final CacheLoader<K, V
     }
   }
 
+  public LoadingCacheDelegate(TypedCache<K, V> cache, CacheConfiguration<K, V> cacheConfig) {
+    this.cache = cache;
+    this.loader = cacheConfig.getLoader();
+    this.cacheName = cacheConfig.getCacheName();
+    this.futureValues = new ConcurrentHashMap<>();
+    this.failOnError = cacheConfig.getFailOnMissingEntries();
+
+    this.duration = cacheConfig.getLoadTimeout();
+    this.timeUnit = cacheConfig.getLoadTimeUnit();
+
+    final MetricFactory metricFactory = cacheConfig.getMetricFactory();
+    if (metricFactory != null) {
+      metricFactory.registerGauge("LoadingCacheDelegate." + cacheName, "mapSize", futureValues::size);
+
+      cacheHits = metricFactory.createCounter("LoadingCacheDelegate." + cacheName, "hits");
+      cacheMiss = metricFactory.createCounter("LoadingCacheDelegate." + cacheName, "miss");
+      cacheErrors = metricFactory.createCounter("LoadingCacheDelegate." + cacheName, "cacheErrors");
+      loaderErrors = metricFactory.createCounter("LoadingCacheDelegate." + cacheName, "loaderErrors");
+      cacheTimeouts = metricFactory.createCounter("LoadingCacheDelegate." + cacheName, "cacheTimeouts");
+      loaderTimeouts = metricFactory.createCounter("LoadingCacheDelegate." + cacheName, "loaderTimeouts");
+
+    } else {
+      cacheHits = null;
+      cacheMiss = null;
+      cacheErrors = null;
+      loaderErrors = null;
+      cacheTimeouts = null;
+      loaderTimeouts = null;
+    }
+  }
+
+
   @Override
   public ComposableFuture<V> getAsync(final K key) {
     return ComposableFutures.build(consumer -> {
@@ -148,7 +181,12 @@ private void fetchFromLoader(final K key, final ComposablePromise<V> promise) {
       loadedResult.consume(loadedRes -> {
         if (loadedRes.isSuccess()) {
           promise.set(loadedRes.getValue());
-          cacheLoadedValue(key, loadedRes);
+          // we are not adding null value into a cache
+          if (loadedRes.getValue() != null) {
+            cacheLoadedValue(key, loadedRes);
+          } else {
+            futureValues.remove(key);
+          }
         } else {
           final Throwable error = loadedRes.getError();
           if (loaderErrors != null) {