@Override public void accept(K key, V value) {
    subject.onNext(Map.entry(key, value));
  }
@Test
  public void multipleKeys() {
    var numberOfEntries = new AtomicInteger();

    // Given this cache...
    var writer = new WriteBehindCacheWriter.Builder<Integer, ZonedDateTime>()
        .writeAction(entries -> numberOfEntries.addAndGet(entries.size()))
        .coalesce(BinaryOperator.maxBy(ZonedDateTime::compareTo))
        .bufferTime(Duration.ofSeconds(1))
        .build();
    Cache<Integer, ZonedDateTime> cache = Caffeine.newBuilder().build();

    // When these cache updates happen...
    for (int i = 1; i <= 3; i++) {
      cache.asMap().computeIfAbsent(i, key -> {
        var value = ZonedDateTime.now();
        writer.accept(key, value);
        return value;
      });
    }

    // Then the write behind action gets 3 entries to write
    await().untilAtomic(numberOfEntries, is(3));
  }