package com.atlassian.vcache.internal.test;
   
   import com.atlassian.utt.concurrency.Barrier;
   import com.atlassian.vcache.ChangeRate;
   import com.atlassian.vcache.DirectExternalCache;
   import com.atlassian.vcache.ExternalCacheSettings;
   import com.atlassian.vcache.ExternalCacheSettingsBuilder;
   import com.atlassian.vcache.PutPolicy;
   import com.atlassian.vcache.StableReadExternalCache;
  import com.google.common.base.Throwables;
  import com.google.common.collect.ImmutableMap;
  import com.google.common.collect.Maps;
  import org.hamcrest.Matchers;
  import org.junit.Assert;
  import org.junit.Before;
  import org.junit.Rule;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.time.Duration;
  import java.util.Map;
  import java.util.Optional;
  import java.util.concurrent.Callable;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.CompletionStage;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  
  import static com.atlassian.vcache.VCacheUtils.unsafeJoin;
  import static com.atlassian.vcache.internal.test.CompletionStageSuccessful.successful;
  import static com.atlassian.vcache.internal.test.CompletionStageSuccessful.successfulWith;
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.hamcrest.Matchers.containsInAnyOrder;
  import static org.hamcrest.Matchers.is;
  import static org.hamcrest.Matchers.not;
  
  /**
   * Base test class for the {@link StableReadExternalCache}.
   */
  public abstract class AbstractStableReadExternalCacheIT {
      private static final Logger log = LoggerFactory.getLogger(AbstractStableReadExternalCacheIT.class);
      private static final String CACHE_NAME = "kids_hobbies";
  
      @Rule
      public LoggingTestWatcher watcher = new LoggingTestWatcher(log);
  
      private StableReadExternalCache<String> cache;
      private DirectExternalCache<String> directCache;
      private final ExecutorService executorService = Executors.newCachedThreadPool();
  
      protected abstract StableReadExternalCache<String> createCache(String name, ExternalCacheSettings settings);
  
      protected abstract DirectExternalCache<String> obtainDirectCache(String name, ExternalCacheSettings settings);
  
      @Before
      public void ensureCache() {
          final ExternalCacheSettings settings = new ExternalCacheSettingsBuilder()
                  .entryGrowthRateHint(ChangeRate.LOW_CHANGE)
                  .entryCountHint(5)
                  .defaultTtl(Duration.ofMinutes(5))
                  .dataChangeRateHint(ChangeRate.HIGH_CHANGE)
                  .build();
          cache = createCache(CACHE_NAME, settings);
  
          directCache = obtainDirectCache(CACHE_NAME, settings);
  
          // Start from a clean slate
          final CompletionStage<Void> rm = directCache.removeAll();
          assertThat(rm, successful());
      }
  
      @Test
      public void single_cache_get_set() throws ExecutionException, InterruptedException {
          final CompletionStage<Optional<String>> eldestGet = cache.get("claira");
  
          assertThat(eldestGet, successfulWith(is(Optional.empty())));
  
          final CompletionStage<Boolean> eldestAdd = cache.put("claira", "dancing", PutPolicy.ADD_ONLY);
  
          assertThat(eldestAdd, successfulWith(is(true)));
  
          final CompletionStage<Optional<String>> eldestGet2 = cache.get("claira");
  
          assertThat(eldestGet2, successfulWith(is(Optional.of("dancing"))));
      }
  
      @Test
      public void dual_cache_get_set() throws ExecutionException, InterruptedException {
          final CompletionStage<Optional<String>> get1 = cache.get("claira");
  
          assertThat(get1, successfulWith(is(Optional.empty())));
  
          // Change will not be visible to main cache
          final CompletionStage<Boolean> dput1 = directCache.put("claira", "dancing", PutPolicy.PUT_ALWAYS);
 
         assertThat(dput1, successfulWith(is(true)));
 
         final CompletionStage<Optional<String>> get2 = cache.get("claira");
 
         assertThat(get2, successfulWith(is(Optional.empty())));
 
         // Add will fail as value exists from the other cache. However, we will now be able to see the
         // value added by the other cache.
         final CompletionStage<Boolean> put1 = cache.put("claira", "singing", PutPolicy.ADD_ONLY);
 
         assertThat(put1, successfulWith(is(false)));
 
         final CompletionStage<Optional<String>> get3 = cache.get("claira");
 
         assertThat(get3, successfulWith(is(Optional.of("dancing"))));
     }
 
     @Test
     public void dual_cache_get_with_supplier() throws ExecutionException, InterruptedException {
         final CompletionStage<Optional<String>> get1 = cache.get("claira");
 
         assertThat(get1, successfulWith(is(Optional.empty())));
 
         final CompletionStage<String> get2 = cache.get("claira", () -> "dancing");
 
         assertThat(get2, successfulWith(is("dancing")));
 
         final CompletionStage<Optional<String>> get3 = cache.get("claira");
 
         assertThat(get3, successfulWith(is(Optional.of("dancing"))));
 
         // Change will not be visible to main cache
         final CompletionStage<Boolean> dput1 = directCache.put("claira", "singing", PutPolicy.PUT_ALWAYS);
         assertThat(dput1, successfulWith(is(true)));
 
         final CompletionStage<Optional<String>> get4 = cache.get("claira");
 
         assertThat(get4, successfulWith(is(Optional.of("dancing"))));
 
         final CompletionStage<String> get5 = cache.get("claira", () -> "riding");
 
         assertThat(get5, successfulWith(is("dancing")));
     }
 
     @Test
     public void dual_cache_get_with_supplier_take2() throws ExecutionException, InterruptedException {
         final CompletionStage<Optional<String>> get1 = cache.get("claira");
 
         assertThat(get1, successfulWith(is(Optional.empty())));
 
         final CompletionStage<Boolean> dput1 = directCache.put("claira", "singing", PutPolicy.PUT_ALWAYS);
         assertThat(dput1, successfulWith(is(true)));
 
         final CompletionStage<String> get2 = cache.get("claira", () -> "riding");
 
         assertThat(get2, successfulWith(is("singing")));
     }
 
     @Test
     public void dual_cache_getBulk() throws ExecutionException, InterruptedException {
         final CompletionStage<Boolean> dput1 = directCache.put("claira", "singing", PutPolicy.PUT_ALWAYS);
         assertThat(dput1, successfulWith(is(true)));
 
         final CompletionStage<Map<String, Optional<String>>> get1 = cache.getBulk("claira", "josephine", "claira");
 
         assertThat(get1, successful());
         final Map<String, Optional<String>> map1 = unsafeJoin(get1);
         assertThat(map1.keySet(), Matchers.containsInAnyOrder("claira", "josephine"));
         assertThat(map1.get("claira"), is(Optional.of("singing")));
         assertThat(map1.get("josephine"), is(Optional.empty()));
 
         final CompletionStage<Boolean> dput2 = directCache.put("josephine", "football", PutPolicy.PUT_ALWAYS);
         assertThat(dput2, successfulWith(is(true)));
 
         final CompletionStage<Boolean> dput3 = directCache.put("jasmin", "skiing", PutPolicy.PUT_ALWAYS);
         assertThat(dput3, successfulWith(is(true)));
 
         final CompletionStage<Map<String, Optional<String>>> get2 = cache.getBulk("claira", "josephine", "jasmin");
 
         assertThat(get2, successful());
         final Map<String, Optional<String>> map2 = unsafeJoin(get2);
         assertThat(map2.keySet(), Matchers.containsInAnyOrder("claira", "josephine", "jasmin"));
         assertThat(map2.get("claira"), is(Optional.of("singing")));
         assertThat(map2.get("josephine"), is(Optional.empty()));
         assertThat(map2.get("jasmin"), is(Optional.of("skiing")));
     }
 
     @Test
     public void remove_normal() throws ExecutionException, InterruptedException {
         final CompletionStage<Boolean> dput1 = directCache.put("claira", "singing", PutPolicy.PUT_ALWAYS);
         assertThat(dput1, successfulWith(is(true)));
 
         final CompletionStage<Boolean> put1 = cache.put("josephine", "football", PutPolicy.PUT_ALWAYS);
         assertThat(put1, successfulWith(is(true)));
 
         final CompletionStage<Map<String, Optional<String>>> get1 = cache.getBulk("claira", "josephine", "jasmin");
 
         assertThat(get1, successful());
         final Map<String, Optional<String>> map1 = unsafeJoin(get1);
         assertThat(map1.keySet(), Matchers.containsInAnyOrder("claira", "josephine", "jasmin"));
         assertThat(map1.get("claira"), is(Optional.of("singing")));
         assertThat(map1.get("josephine"), is(Optional.of("football")));
         assertThat(map1.get("jasmin"), is(Optional.empty()));
 
         final CompletionStage<Void> rm1 = cache.remove("jasmin", "claira", "claira", "josephine");
 
         assertThat(rm1, successful());
 
         final CompletionStage<Map<String, Optional<String>>> dget1 = directCache.getBulk("claira", "josephine", "jasmin");
 
         assertThat(dget1, successful());
         final Map<String, Optional<String>> dmap1 = unsafeJoin(dget1);
         assertThat(dmap1.keySet(), Matchers.containsInAnyOrder("claira", "josephine", "jasmin"));
         assertThat(dmap1.get("claira"), is(Optional.empty()));
         assertThat(dmap1.get("josephine"), is(Optional.empty()));
         assertThat(dmap1.get("jasmin"), is(Optional.empty()));
 
         final CompletionStage<Map<String, Optional<String>>> get2 = cache.getBulk("claira", "josephine", "jasmin");
 
         assertThat(get2, successful());
         final Map<String, Optional<String>> map2 = unsafeJoin(get2);
         assertThat(map2.keySet(), Matchers.containsInAnyOrder("claira", "josephine", "jasmin"));
         assertThat(map2.get("claira"), is(Optional.empty()));
         assertThat(map2.get("josephine"), is(Optional.empty()));
         assertThat(map2.get("jasmin"), is(Optional.empty()));
     }
 
     @Test
     public void simple_removeAll() throws ExecutionException, InterruptedException {
         final CompletionStage<Boolean> put1 = cache.put("claira", "dancing", PutPolicy.PUT_ALWAYS);
 
         assertThat(put1, successfulWith(is(true)));
 
         final CompletionStage<Optional<String>> get1 = cache.get("claira");
 
         assertThat(get1, successfulWith(is(Optional.of("dancing"))));
 
         final CompletionStage<Void> rm1 = cache.removeAll();
 
         assertThat(rm1, successful());
 
         final CompletionStage<Optional<String>> get2 = cache.get("claira");
 
         assertThat(get2, successfulWith(is(Optional.empty())));
     }
 
     @Test
     public void simple_getBulk_function() throws ExecutionException, InterruptedException {
         final CompletionStage<Map<String, String>> get1 = cache.getBulk(
                 keys -> Maps.asMap(keys, k -> k + "-1"), "claira");
 
         assertThat(get1, successful());
         assertThat(unsafeJoin(get1).keySet(), containsInAnyOrder("claira"));
         assertThat(unsafeJoin(get1).values(), containsInAnyOrder("claira-1"));
 
         final CompletionStage<Map<String, String>> get2 = cache.getBulk(
                 keys -> Maps.asMap(keys, k -> k + "-2"), "claira", "josephine", "claira");
 
         assertThat(get2, successful());
         assertThat(unsafeJoin(get2).keySet(), containsInAnyOrder("claira", "josephine"));
         assertThat(unsafeJoin(get2).values(), containsInAnyOrder("claira-1", "josephine-2"));
     }
 
     @Test
     public void dual_getBulk_function() throws ExecutionException, InterruptedException {
         final CompletionStage<Map<String, String>> get1 = cache.getBulk(
                 keys -> Maps.asMap(keys, k -> k + "-1"), "claira");
 
         assertThat(get1, successful());
         assertThat(unsafeJoin(get1).keySet(), containsInAnyOrder("claira"));
         assertThat(unsafeJoin(get1).values(), containsInAnyOrder("claira-1"));
 
         final CompletionStage<Boolean> mput1 = directCache.put("claira", "singing", PutPolicy.PUT_ALWAYS);
         assertThat(mput1, successfulWith(is(true)));
 
         final CompletionStage<Map<String, String>> get2 = cache.getBulk(
                 keys -> Maps.asMap(keys, k -> k + "-2"), "claira", "josephine", "claira");
 
         assertThat(get2, successful());
         assertThat(unsafeJoin(get2).keySet(), containsInAnyOrder("claira", "josephine"));
         assertThat(unsafeJoin(get2).values(), containsInAnyOrder("claira-1", "josephine-2"));
 
         final CompletionStage<Void> rm1 = cache.remove("claira");
 
         assertThat(rm1, successful());
 
         final CompletionStage<Map<String, String>> get3 = cache.getBulk(
                 keys -> Maps.asMap(keys, k -> k + "-3"), "claira", "josephine", "jasmin");
 
         assertThat(get3, successful());
         assertThat(unsafeJoin(get3).keySet(), containsInAnyOrder("claira", "josephine", "jasmin"));
         assertThat(unsafeJoin(get3).values(), containsInAnyOrder("claira-3", "josephine-2", "jasmin-3"));
     }
 
     @SuppressWarnings("ConstantConditions")
     @Test
     public void check_null_detection() {
         assertThat(cache.get("kenny", () -> null), not(successful()));
         assertThat(cache.put("key", null, PutPolicy.ADD_ONLY), not(successful()));
         assertThat(cache.getBulk(
                 strings -> strings.stream().collect(Collectors.toMap(k -> k, k -> null)),
                 "extra"),
                 not(successful()));
     }
 
     @Test
     public void both_getBulk_and_get_fail() throws InterruptedException {
         //Scenario:
         // T1: bulkGet A, B, C
         // T2: get B, supplier
         // T1: bulkGet completes load with an exception
         // T2: get B, supplier -> fails with another exception
 
         final Barrier blockedInFactory = new Barrier();
         final Barrier resumeInFactory = new Barrier();
 
         final RuntimeException exBulk = new RuntimeException("getBulk() exception");
         final CompletableFuture<Map<String, String>> t1Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
             log.info("{}: about to wait", Thread.currentThread().getName());
             blockedInFactory.signal();
             resumeInFactory.await();
             log.info("{}: About to throw exception", Thread.currentThread().getName());
             throw exBulk;
         }, "A", "B", "C").toCompletableFuture().join());
 
         final RuntimeException exSingle = new RuntimeException("get() exception");
         blockedInFactory.await();
         final CompletableFuture<String> t2Result = runAndWaitForStart(() -> cache.get("B", () -> {
             log.info("{}: About to throw exception", Thread.currentThread().getName());
             throw exSingle;
         }).toCompletableFuture().join());
 
         log.info("{}: About to signal on barrierBulk", Thread.currentThread().getName());
         resumeInFactory.signal(); //Wait until T2 cache.get calls join
 
         Assert.assertThat(t1Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exBulk));
         Assert.assertThat(t2Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exSingle));
         Assert.assertThat(cache.get("A").toCompletableFuture().join(), is(Optional.empty()));
         Assert.assertThat(cache.get("B").toCompletableFuture().join(), is(Optional.empty()));
         Assert.assertThat(cache.get("C").toCompletableFuture().join(), is(Optional.empty()));
     }
 
     @Test
     public void both_get_and_getBulk_fail() throws InterruptedException {
         //Scenario:
         // T1: get B, supplier
         // T2: bulkGet A, B, C
         // T1: get B, supplier -> fails with another exception
         // T2: bulkGet completes load with an exception
 
         final Barrier blockedInFactory = new Barrier();
         final Barrier resumeInFactory = new Barrier();
 
         final RuntimeException exSingle = new RuntimeException("get() exception");
         final CompletableFuture<String> t1Result = runAndWaitForStart(() -> cache.get("B", () -> {
             log.info("{}: About to throw exception", Thread.currentThread().getName());
             blockedInFactory.signal();
             resumeInFactory.await();
             throw exSingle;
         }).toCompletableFuture().join());
 
         final RuntimeException exBulk = new RuntimeException("getBulk() exception");
         blockedInFactory.await();
         final CompletableFuture<Map<String, String>> t2Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
             log.info("{}: about to wait", Thread.currentThread().getName());
             log.info("{}: About to throw exception", Thread.currentThread().getName());
             throw exBulk;
         }, "A", "B", "C").toCompletableFuture().join());
 
         log.info("{}: About to signal on barrierBulk", Thread.currentThread().getName());
         resumeInFactory.signal(); //Wait until T2 cache.get calls join
 
         Assert.assertThat(t1Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exSingle));
         Assert.assertThat(t2Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exBulk));
         Assert.assertThat(cache.get("A").toCompletableFuture().join(), is(Optional.empty()));
         Assert.assertThat(cache.get("B").toCompletableFuture().join(), is(Optional.empty()));
         Assert.assertThat(cache.get("C").toCompletableFuture().join(), is(Optional.empty()));
     }
 
     @Test
     public void get_fails_and_getBulk_succeeds() throws InterruptedException {
         //Scenario:
         // T1: get B, supplier, fails
         // T2: bulkGet A, B, C succeeds
         // T1: get B, supplier -> fails with another exception
         // T2: bulkGet completes with a map
 
         final Barrier blockedInFactory = new Barrier();
         final Barrier resumeInFactory = new Barrier();
 
         final RuntimeException exSingle = new RuntimeException("get() exception");
         final CompletableFuture<String> t1Result = runAndWaitForStart(() -> cache.get("B", () -> {
             log.info("{}: about to wait", Thread.currentThread().getName());
             blockedInFactory.signal();
             resumeInFactory.await();
             log.info("{}: About to throw exception", Thread.currentThread().getName());
             throw exSingle;
         }).toCompletableFuture().join());
 
         blockedInFactory.await();
         final CompletableFuture<Map<String, String>> t2Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
             log.info("{}: About to return numbers", Thread.currentThread().getName());
             return ImmutableMap.of("A", "1", "B", "2", "C", "3");
         }, "A", "B", "C").toCompletableFuture().join());
 
         log.info("{}: About to signal on barrierBulk", Thread.currentThread().getName());
         resumeInFactory.signal(); //Wait until T2 cache.get calls join
 
         Assert.assertThat(t1Result.handle((r, e) -> {
             if (e != null) {
                 return Throwables.getRootCause(e);
             } else {
                 log.info("r: {}", r);
                 // This forces a fail.
                 return new RuntimeException();
             }
         }).join(), is(exSingle));
 
         Assert.assertThat(t2Result.join().get("A"), is("1"));
         Assert.assertThat(t2Result.join().get("B"), is("2"));
         Assert.assertThat(t2Result.join().get("C"), is("3"));
 
         Assert.assertThat(cache.get("A").toCompletableFuture().join(), is(Optional.of("1")));
         Assert.assertThat(cache.get("B").toCompletableFuture().join(), is(Optional.of("2")));
         Assert.assertThat(cache.get("C").toCompletableFuture().join(), is(Optional.of("3")));
     }
 
     @Test
     public void getBulk_fail_and_get_succeed() throws InterruptedException {
         //Scenario:
         // T1: bulkGet A, B, C
         // T2: get B, supplier
         // T1: bulkGet function throws exception
         // T2: get B, supplier -> returns value
 
         final Barrier blockedInFactory = new Barrier();
         final Barrier resumeInFactory = new Barrier();
 
         final RuntimeException exBulk = new RuntimeException("getBulk() exception");
         final CompletableFuture<Map<String, String>> t1Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
             log.info("{}: about to wait", Thread.currentThread().getName());
             blockedInFactory.signal();
             resumeInFactory.await();
             log.info("{}: About to throw exception", Thread.currentThread().getName());
             throw exBulk;
         }, "A", "B", "C").toCompletableFuture().join());
 
         blockedInFactory.await();
         final CompletableFuture<String> t2Result = runAndWaitForStart(() -> cache.get("B", () -> "bee").toCompletableFuture().join());
 
         log.info("{}: About to signal on barrierBulk", Thread.currentThread().getName());
         resumeInFactory.signal(); //Wait until T2 cache.get calls join
 
 
         Assert.assertThat(t1Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exBulk));
         Assert.assertThat(t2Result.join(), is("bee"));
         Assert.assertThat(cache.get("A").toCompletableFuture().join(), is(Optional.empty()));
         Assert.assertThat(cache.get("B").toCompletableFuture().join(), is(Optional.of("bee")));
         Assert.assertThat(cache.get("C").toCompletableFuture().join(), is(Optional.empty()));
     }
 
     @Test
     public void getBulk_pass_and_get_reuse() throws InterruptedException {
         //Scenario:
         // T1: bulkGet A, B, C
         // T2: get B, supplier
         // T1: bulkGet function returns values
         // T2: get B, supplier -> returns bulkGet value
 
         final Barrier blockedInFactory = new Barrier();
         final Barrier resumeInFactory = new Barrier();
 
         final CompletableFuture<Map<String, String>> t1Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
             log.info("{}: About to await on barrierBulk", Thread.currentThread().getName());
             blockedInFactory.signal();
             resumeInFactory.await();
             log.info("{}: resuming", Thread.currentThread().getName());
             return keys.stream().collect(Collectors.toMap(k -> k, k -> k + "-1"));
         }, "A", "B", "C").toCompletableFuture().join());
 
         blockedInFactory.await();
         final CompletableFuture<String> t2Result = runAndWaitForStart(() -> cache.get("B", () -> "bee").toCompletableFuture().join());
 
         log.info("{}: About to signal on barrierBulk", Thread.currentThread().getName());
         resumeInFactory.signal(); //Wait until T2 cache.get calls join
         log.info("{}: and now", Thread.currentThread().getName());
 
 
         final Map<String, String> t1ResultData = t1Result.join();
         Assert.assertThat(t1ResultData.get("A"), is("A-1"));
         Assert.assertThat(t1ResultData.get("B"), is("B-1"));
         Assert.assertThat(t1ResultData.get("C"), is("C-1"));
         Assert.assertThat(t2Result.join(), is("B-1"));
         Assert.assertThat(cache.get("A").toCompletableFuture().join(), is(Optional.of("A-1")));
         Assert.assertThat(cache.get("B").toCompletableFuture().join(), is(Optional.of("B-1")));
         Assert.assertThat(cache.get("C").toCompletableFuture().join(), is(Optional.of("C-1")));
     }
 
     @Test
     public void potential_deadlock() {
         final CompletableFuture<String> get = cache.get("lockType", () -> {
             // People have actually done this!!
             cache.remove("lockType");
             return "lockwood";
         }).toCompletableFuture();
 
         assertThat(get.isCompletedExceptionally(), is(true));
     }
 
     @Test
     public void repeated_concurrency() {
         IntStream.range(1, 50).forEach((i) -> {
             try {
                 getBulk_fail_and_get_succeed();
                 cache.removeAll();
                 getBulk_pass_and_get_reuse();
                 cache.removeAll();
                 get_fails_and_getBulk_succeeds();
                 cache.removeAll();
                 both_get_and_getBulk_fail();
                 cache.removeAll();
                 both_getBulk_and_get_fail();
                 cache.removeAll();
 
             } catch (InterruptedException e) {
                 throw new RuntimeException(e);
             }
         });
     }
 
     private <T> CompletableFuture<T> runAndWaitForStart(final Callable<T> r) {
         final Barrier barrier = new Barrier();
         log.info("{}: About to submit()", Thread.currentThread().getName());
         final Future<T> result = executorService.submit(() -> {
             log.info("{}: About to signal()", Thread.currentThread().getName());
             barrier.signal();
             log.info("{}: resume after signal", Thread.currentThread().getName());
             return r.call();
         });
 
         log.info("{}: About to await()", Thread.currentThread().getName());
         barrier.await();
         log.info("{}: resume after await()", Thread.currentThread().getName());
 
         return CompletableFuture.supplyAsync(() -> {
             try {
                 return result.get();
             } catch (InterruptedException | ExecutionException e) {
                 throw new RuntimeException(e);
             }
         });
     }
 }