Caffeine Caching Library

An IndexedCache allows you to associate a single cache value with multiple unique keys (similar to a relational database with primary and secondary indexes). This enables fast retrieval of the same object using different identifiers (e.g., looking up a user by ID, email, or username).

When a value is updated, deleted, or evicted, the IndexedCache ensures that all associated key mappings are kept consistent.

To implement this, you use an IndexedCache.Builder to define:

1. A primary key function to identify the canonical key for the value.
2. Secondary key functions to define the alternative lookup paths.
3. A data loader function to fetch the value on a cache miss.
4. Standard Caffeine bounding constraints like expireAfterWrite or maximumSize.

var cache = new IndexedCache.Builder<UserKey, User>()
    .primaryKey(user -> new UserById(user.id()))
    .addSecondaryKey(user -> new UserByEmail(user.email()))
    .addSecondaryKey(user -> new UserByLogin(user.username()))
    .expireAfterWrite(Duration.ofMinutes(5))
    .maximumSize(10_000)
    .build(this::findUser);

// All these calls return the same instance if the user is already cached
var userByEmail = cache.get(new UserByEmail("john.doe@example.com"));
var userByLogin = cache.get(new UserByLogin("john.doe"));

If all loads (not just refreshes) should be collected into batches, implement AsyncCacheLoader. This is most suitable for an AsyncLoadingCache because it does not block map operations while an entry is being loaded. All requests are submitted to a Reactor Sink via asyncLoad and processed in batches by a subscriber.

public final class CoalescingBulkLoader<K, V> implements AsyncCacheLoader<K, V> {
  private final Function<Set<K>, Map<K, V>> mappingFunction;
  private final Sinks.Many<Request<K, V>> sink;

  public CoalescingBulkLoader(int maxSize, Duration maxTime, int parallelism, 
      Function<Set<K>, Map<K, V>> mappingFunction) {
    this.sink = Sinks.many().unicast().onBackpressureBuffer();
    this.mappingFunction = requireNonNull(mappingFunction);
    sink.asFlux()
        .bufferTimeout(maxSize, maxTime)
        .map(requests -> requests.stream().collect(
            toMap(Entry::getKey, Entry::getValue)))
        .parallel(parallelism)
        .runOn(Schedulers.boundedElastic())
        .subscribe(this::handle);
  }

  @Override public synchronized CompletableFuture<V> asyncLoad(K key, Executor e) {
    var entry = Map.entry(key, new CompletableFuture<V>());
    sink.tryEmitNext(entry).orThrow();
    return entry.getValue();
  }

  private void handle(Map<K, CompletableFuture<V>> requests) {
    try {
      var results = mappingFunction.apply(requests.keySet());
      requests.forEach((key, result) -> result.complete(results.get(key)));
    } catch (Throwable t) {
      requests.forEach((key, result) -> result.completeExceptionally(t));
    }
  }
}

To aggregate only background refreshes while keeping explicit loads immediate, implement CacheLoader. In this pattern, load and loadAll invoke the mapping function directly (synchronously), while asyncReload submits the request to a Reactor Sink for batching.

Note: The asyncReload method must be synchronized because the Reactor Sink does not support concurrent submissions.

public final class CoalescingBulkLoader<K, V> implements CacheLoader<K, V> {
  private final Function<Set<K>, Map<K, V>> mappingFunction;
  private final Sinks.Many<Request<K, V>> sink;

  /**
   * @param maxSize the maximum entries to collect before performing a bulk request
   * @param maxTime the maximum duration to wait before performing a bulk request
   * @param parallelism the number of parallel bulk loads that can be performed
   * @param mappingFunction the function to compute the values
   */
  public CoalescingBulkLoader(int maxSize, Duration maxTime, int parallelism, 
      Function<Set<K>, Map<K, V>> mappingFunction) {
    this.sink = Sinks.many().unicast().onBackpressureBuffer();
    this.mappingFunction = requireNonNull(mappingFunction);
    sink.asFlux()
        .bufferTimeout(maxSize, maxTime)
        .map(requests -> requests.stream().collect(
            toMap(Entry::getKey, Entry::getValue)))
        .parallel(parallelism)
        .runOn(Schedulers.boundedElastic())
        .subscribe(this::handle);
  }

  @Override public V load(K key) {
    return loadAll(Set.of(key)).get(key);
  }

  @Override public Map<K, V> loadAll(Set<? extends K> keys) {
    return mappingFunction.apply(keys);
  }

  @Override public synchronized CompletableFuture<V> asyncReload(K key, V oldValue, Executor e) {
    var entry = Map.entry(key, new CompletableFuture<V>());
    sink.tryEmitNext(entry).orThrow();
    return entry.getValue();
  }

  private void handle(Map<K, CompletableFuture<V>> requests) {
    try {
      var results = mappingFunction.apply(requests.keySet());
      requests.forEach((key, result) -> result.complete(results.get(key)));
    } catch (Throwable t) {
      requests.forEach((key, result) -> result.completeExceptionally(t));
    }
  }
}

You can use Failsafe's RetryPolicy to make Caffeine cache operations resilient to intermittent failures. There are two primary ways to apply retries:

1. Retry outside the cache loader (Synchronous): Wrap the cache.get call itself. This retries the entire cache lookup/load process.
2. Retry inside the cache load (Asynchronous): For AsyncCache, wrap the loading function inside failsafe.getAsync. This retries only the specific loading operation if it fails, without re-triggering the cache lookup logic.

Note: This requires the Failsafe library.

var retryPolicy = RetryPolicy.builder()
    .withDelay(Duration.ofSeconds(1))
    .withMaxAttempts(3)
    .build();
var failsafe = Failsafe.with(retryPolicy);

// Retry outside of the cache loader for synchronous calls
Cache<K, V> cache = Caffeine.newBuilder().build();
failsafe.get(() -> cache.get(key, key -> /* intermittent failures */ ));

// Optionally, retry inside the cache load for asynchronous calls
AsyncCache<K, V> asyncCache = Caffeine.newBuilder().buildAsync();
asyncCache.get(key, (key, executor) -> failsafe.getAsync(() -> /* intermittent failure */ ));