@Override public void accept(K key, V value) {
    subject.onNext(Map.entry(key, value));
  }
@Test
  public void singleKey_mostRecent() {
    var timeInWriteBehind = new AtomicReference<ZonedDateTime>();
    var numberOfEntries = new AtomicInteger();

    // Given this cache...
    var writer = new WriteBehindCacheWriter.Builder<Long, ZonedDateTime>()
        .coalesce(BinaryOperator.maxBy(ZonedDateTime::compareTo))
        .bufferTime(Duration.ofSeconds(1))
        .writeAction(entries -> {
          // We might get here before the cache has been written to,
          // so just wait for the next time we are called
          if (entries.isEmpty()) {
            return;
          }

          var zonedDateTime = entries.values().iterator().next();
          timeInWriteBehind.set(zonedDateTime);
          numberOfEntries.set(entries.size());
        }).build();
    Cache<Long, ZonedDateTime> cache = Caffeine.newBuilder().build();

    // When these cache updates happen...
    var latest = ZonedDateTime.now().truncatedTo(DAYS);
    for (int i = 0; i < 4; i++) {
      latest = latest.plusNanos(200);

      var value = latest;
      cache.asMap().compute(1L, (key, oldValue) -> {
        writer.accept(key, value);
        return value;
      });
    }

    // Then the write behind action gets 1 entry to write with the most recent time
    await().untilAtomic(numberOfEntries, is(1));
    await().untilAtomic(timeInWriteBehind, is(latest));
  }