@Override
public void deleteAll(Iterable<? extends K> keys) throws BulkCacheWritingException {
  if (cacheWriter != null) {
    Set<K> allKeys = new HashSet<>();
    for (K key : keys) {
      allKeys.add(key);
    }

    try {
      cacheWriter.deleteAll(allKeys);
    } catch (Exception e) {
      // the remaining keys were not written per 107 spec
      Map<?, Exception> failures = failures(allKeys, e);
      Set<?> successes = successes(keys, failures.keySet());
      throw new BulkCacheWritingException(failures, successes);
    }
  }
}
 

private void cacheLoaderWriterWriteAllCall(Iterable<? extends Map.Entry<? extends K, ? extends V>> entries, Map<K, V> entriesToRemap, Set<K> successes, Map<K, Exception> failures) throws IllegalStateException {
  Map<K, V> toWrite = new HashMap<>();
  for (Map.Entry<? extends K, ? extends V> entry: entries) {
    V value = entriesToRemap.get(entry.getKey());
    if (value == null) {
      continue;
    }

    toWrite.put(entry.getKey(), value);
  }
  try {
    if (! toWrite.isEmpty()) {
      // write all entries of this batch
      cacheLoaderWriter.writeAll(toWrite.entrySet());
      successes.addAll(toWrite.keySet());
    }
  } catch (BulkCacheWritingException bcwe) {
    collectSuccessesAndFailures(bcwe, successes, failures);
  } catch (Exception e) {
    for (K key: toWrite.keySet()) {
      failures.put(key, e);
    }
  }
}
 

/**
 * {@inheritDoc}
 * <p>
 * If this method throws an exception <i>other</i> than a
 * {@link BulkCacheWritingException BulkCacheWritingException}, the
 * cache image maintained by this {@code CacheLoaderWriter} is in an inconsistent state.
 */
@Override
public void writeAll(final Iterable<? extends Map.Entry<? extends String, ? extends String>> entries)
    throws Exception {

  final Set<String> successes = new LinkedHashSet<>();
  final Map<String, Exception> failures = new LinkedHashMap<>();

  for (final Entry<? extends String, ? extends String> entry : entries) {
    final String key = entry.getKey();
    if (key.equals(this.completeFailureKey)) {
      throw new CompleteFailureException();
    }
    try {
      this.write(key, entry.getValue());
      successes.add(key);
    } catch (Exception e) {
      //noinspection ThrowableResultOfMethodCallIgnored
      failures.put(key, e);
    }
  }

  if (!failures.isEmpty()) {
    throw new BulkCacheWritingException(failures, successes);
  }
}
 

/**
 * {@inheritDoc}
 * <p>
 * If this method throws an exception <i>other</i> than a
 * {@link BulkCacheWritingException BulkCacheWritingException}, the
 * cache image maintained by this {@code CacheLoaderWriter} is in an inconsistent state.
 */
@Override
public void deleteAll(final Iterable<? extends String> keys) throws Exception {
  final Set<String> successes = new LinkedHashSet<>();
  final Map<String, Exception> failures = new LinkedHashMap<>();

  for (final String key : keys) {
    if (key.equals(this.completeFailureKey)) {
      throw new CompleteFailureException();
    }
    try {
      this.delete(key);
      successes.add(key);
    } catch (Exception e) {
      //noinspection ThrowableResultOfMethodCallIgnored
      failures.put(key, e);
    }
  }

  if (!failures.isEmpty()) {
    throw new BulkCacheWritingException(failures, successes);
  }
}
 

@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testRemoveAllWithWriterException() throws Exception {
  doAnswer(invocation -> {
    Iterable<Integer> iterable = (Iterable) invocation.getArguments()[0];
    Set<Integer> result = new HashSet<>();

    for (Integer i : iterable) {
      switch (i) {
        case 2:
          throw new Exception("Mock Exception: cannot write 2");
        case 1:
        case 3:
        case 4:
          result.add(i);
          break;
        default:
          throw new AssertionError("should not try to delete key " + i);
      }
    }

    return result;
  }).when(cacheLoaderWriter).deleteAll(ArgumentMatchers.<Iterable>any());

  try {
    testCache.removeAll(new HashSet<Number>(Arrays.asList(1, 2, 3, 4)));
    fail("expected CacheWritingException");
  } catch (BulkCacheWritingException ex) {
    assertThat(ex.getFailures().size(), is(1));
    assertThat(ex.getFailures().get(2), is(notNullValue()));
    assertThat(ex.getSuccesses().size(), is(3));
    assertThat(ex.getSuccesses().containsAll(Arrays.asList(1, 3, 4)), is(true));
  }
}
 

@Test
public void testPutAll_with_cache_writer_that_throws_exception() throws Exception {
  CacheConfigurationBuilder cacheConfigurationBuilder = CacheConfigurationBuilder.newCacheConfigurationBuilder(String.class, String.class,
      heap(100));

  CacheLoaderWriterProvider cacheLoaderWriterProvider = mock(CacheLoaderWriterProvider.class);
  CacheLoaderWriter cacheLoaderWriterThatThrows = mock(CacheLoaderWriter.class);
  doThrow(new RuntimeException("We should not have called .write() but .writeAll()")).when(cacheLoaderWriterThatThrows).write(ArgumentMatchers
      .any(), ArgumentMatchers.any());
  doThrow(new Exception("Simulating an exception from the cache writer")).when(cacheLoaderWriterThatThrows).writeAll(ArgumentMatchers.any(Iterable.class));
  when(cacheLoaderWriterProvider.createCacheLoaderWriter(anyString(), ArgumentMatchers.any(CacheConfiguration.class))).thenReturn(cacheLoaderWriterThatThrows);

  CacheManagerBuilder<CacheManager> managerBuilder = CacheManagerBuilder.newCacheManagerBuilder().using(cacheLoaderWriterProvider);
  CacheConfiguration<String, String> cacheConfiguration = cacheConfigurationBuilder.withLoaderWriter(cacheLoaderWriterThatThrows).build();
  CacheManager cacheManager = managerBuilder.withCache("myCache", cacheConfiguration).build(true);

  Cache<String, String> myCache = cacheManager.getCache("myCache", String.class, String.class);

  HashMap<String, String> stringStringHashMap = new HashMap<>();
  for (int i = 0; i < 3; i++) {
    stringStringHashMap.put("key" + i, "value" + i);
  }

  // the call to putAll
  try {
    myCache.putAll(stringStringHashMap);
    fail();
  } catch (BulkCacheWritingException bcwe) {
    assertThat(bcwe.getFailures().size(), is(3));
    assertThat(bcwe.getSuccesses().size(), is(0));
  }

}
 

@Override
public void writeAll(final Iterable<? extends Map.Entry<? extends Long, ? extends byte[]>> iterable) throws BulkCacheWritingException, Exception {
  throw new UnsupportedOperationException();
}
 

@Override
public void deleteAll(final Iterable<? extends Long> iterable) throws BulkCacheWritingException, Exception {
  throw new UnsupportedOperationException();
}
 

@Test
public void testRemoveAll_cache_writer_throws_exception() throws Exception {
  CacheConfigurationBuilder cacheConfigurationBuilder = CacheConfigurationBuilder.newCacheConfigurationBuilder(String.class, String.class,
      heap(100));

  CacheLoaderWriterProvider cacheLoaderWriterProvider = mock(CacheLoaderWriterProvider.class);
  CacheLoaderWriter cacheLoaderWriterThatThrows = mock(CacheLoaderWriter.class);
  doThrow(new Exception("Simulating an exception from the cache writer")).when(cacheLoaderWriterThatThrows).deleteAll(ArgumentMatchers.any(Iterable.class));
  when(cacheLoaderWriterProvider.createCacheLoaderWriter(anyString(), ArgumentMatchers.any(CacheConfiguration.class))).thenReturn(cacheLoaderWriterThatThrows);

  CacheManagerBuilder<CacheManager> managerBuilder = CacheManagerBuilder.newCacheManagerBuilder().using(cacheLoaderWriterProvider);
  CacheConfiguration<String, String> cacheConfiguration = cacheConfigurationBuilder.withLoaderWriter(cacheLoaderWriterThatThrows)
          .build();
  CacheManager cacheManager = managerBuilder.withCache("myCache", cacheConfiguration).build(true);

  Cache<String, String> myCache = cacheManager.getCache("myCache", String.class, String.class);

  for (int i = 0; i < 3; i++) {
    myCache.put("key" + i, "value" + i);
  }

  doThrow(new RuntimeException("We should not have called .write() but .writeAll()")).when(cacheLoaderWriterThatThrows).write(ArgumentMatchers
      .any(), ArgumentMatchers.any());

  Set<String> fewKeysSet = new HashSet<String>() {
    {
      add("key0");
      add("key2");
    }
  };

  // the call to removeAll
  try {
    myCache.removeAll(fewKeysSet);
    fail();
  } catch (BulkCacheWritingException bcwe) {
    assertThat(bcwe.getFailures().size(), is(2));
    assertThat(bcwe.getSuccesses().size(), is(0));
  }

}
 

@Override
public void writeAll(Iterable<? extends Map.Entry<? extends K, ? extends V>> entries) throws BulkCacheWritingException, Exception {
  delegate.writeAll(entries);
}
 

@Override
public void deleteAll(Iterable<? extends K> keys) throws BulkCacheWritingException, Exception {
  delegate.deleteAll(keys);
}
 

public void performOperation(CacheLoaderWriter<K, V> cacheLoaderWriter) throws BulkCacheWritingException, Exception {
  cacheLoaderWriter.writeAll(entries);
}
 

/**
 * {@inheritDoc}
 */
public void performOperation(CacheLoaderWriter<K, V> cacheLoaderWriter) throws BulkCacheWritingException, Exception {
  cacheLoaderWriter.deleteAll(entries);
}
 

private Map<K, ValueHolder<V>> getkValueHolderMap(Ehcache.PutAllFunction<K, V> remappingFunction, Set<K> successes, Map<K, Exception> failures) throws StoreAccessException {
  // Copy all entries to write into a Map
  Ehcache.PutAllFunction<K, V> putAllFunction = remappingFunction;
  Map<K, V> entriesToRemap = CollectionUtil.copyMapButFailOnNull(putAllFunction.getEntriesToRemap());

  int[] actualPutCount = {0};

  // The compute function that will return the keys to their NEW values, taking the keys to their old values as input;
  // but this could happen in batches, i.e. not necessary containing all of the entries of the Iterable passed to this method
  Function<Iterable<? extends Map.Entry<? extends K, ? extends V>>, Iterable<? extends Map.Entry<? extends K, ? extends V>>> computeFunction =
          entries1 -> {
            // If we have a writer, first write this batch
            cacheLoaderWriterWriteAllCall(entries1, entriesToRemap, successes, failures);

            int size = CollectionUtil.findBestCollectionSize(entries1, 1);
            Map<K, V> mutations = new LinkedHashMap<>(size);

            // then record we handled these mappings
            for (Map.Entry<? extends K, ? extends V> entry : entries1) {
              K key = entry.getKey();
              V existingValue = entry.getValue();
              V newValue = entriesToRemap.remove(key);

              if (newValueAlreadyExpired(LOG, expiry, key, existingValue, newValue)) {
                mutations.put(key, null);
              } else if (successes.contains(key)) {
                ++actualPutCount[0];
                mutations.put(key, newValue);

              } else {
                mutations.put(key, existingValue);
              }
            }

            // Finally return the values to be installed in the Cache's Store
            return mutations.entrySet();
          };

  Map<K, ValueHolder<V>> computedMap = delegate.bulkCompute(putAllFunction.getEntriesToRemap().keySet(), computeFunction);
  if (!failures.isEmpty()) {
    throw new BulkCacheWritingException(failures, successes);
  }
  return computedMap;
}
 

@SuppressWarnings({ "unchecked" })
private static <K> void collectSuccessesAndFailures(BulkCacheWritingException bcwe, Set<K> successes, Map<K, Exception> failures) {
  successes.addAll((Collection<K>)bcwe.getSuccesses());
  failures.putAll((Map<K, Exception>)bcwe.getFailures());
}
 

@Override
public void putAll(Map<? extends K, ? extends V> entries) throws BulkCacheWritingException {
  throw new UnsupportedOperationException("Implement me!");
}
 

/**
 * {@inheritDoc}
 */
@Override
public void putAll(Map<? extends K, ? extends V> entries) throws BulkCacheWritingException {
  cache.putAll(entries);
}
 

/**
 * {@inheritDoc}
 */
@Override
public void removeAll(Set<? extends K> keys) throws BulkCacheWritingException {
  cache.removeAll(keys);
}
 

protected void doRemoveAll(final Set<? extends K> keys) throws BulkCacheWritingException, StoreAccessException {
  RemoveAllFunction<K, V> removeAllFunction = new RemoveAllFunction<>();
  store.bulkCompute(keys, removeAllFunction);
  addBulkMethodEntriesCount(BulkOps.REMOVE_ALL, removeAllFunction.getActualRemoveCount().get());
}
 

/**
 * Associates all the provided key:value pairs.
 *
 * @param entries key:value pairs to associate, keys or values may not be {@code null}
 *
 * @throws NullPointerException if the {@code Map} or any of the contained keys or values are {@code null}.
 * @throws BulkCacheWritingException if the {@link CacheLoaderWriter} associated with this cache threw an
 * {@link Exception} while writing given key:value pairs to the underlying system of record.
 */
void putAll(Map<? extends K, ? extends V> entries) throws BulkCacheWritingException;
 

/**
 * Removes any associated value for the given key set.
 *
 * @param keys keys to remove values for, may not be {@code null}
 *
 * @throws NullPointerException if the {@code Set} or any of the contained keys are {@code null}.
 * @throws BulkCacheWritingException if the {@link CacheLoaderWriter} associated with this cache threw an
 * {@link Exception} while removing mappings for given keys from the underlying system of record.
 */
void removeAll(Set<? extends K> keys) throws BulkCacheWritingException;
 

/**
 * Perform the batch operation for a particular batch writer
 *
 */
void performOperation(CacheLoaderWriter<K, V> cacheLoaderWriter) throws BulkCacheWritingException, Exception;
 

protected abstract void doPutAll(Map<? extends K, ? extends V> entries) throws StoreAccessException, BulkCacheWritingException; 

protected abstract void doRemoveAll(Set<? extends K> keys) throws BulkCacheWritingException, StoreAccessException;