public <T> CompletableFuture<T> withLockAsync(final List<String> e164s,
      final Supplier<CompletableFuture<T>> taskSupplier,
      final Executor executor) {

    if (e164s.isEmpty()) {
      throw new IllegalArgumentException("List of e164s to lock must not be empty");
    }

    final List<LockItem> lockItems = new ArrayList<>(e164s.size());

    return CompletableFuture.runAsync(() -> {
          for (final String e164 : e164s) {
            try {
              lockItems.add(lockClient.acquireLock(AcquireLockOptions.builder(e164)
                  .withAcquireReleasedLocksConsistently(true)
                  .build()));
            } catch (final InterruptedException e) {
              throw new CompletionException(e);
            }
          }
        }, executor)
        .thenCompose(ignored -> taskSupplier.get())
        .whenCompleteAsync((ignored, throwable) -> lockItems.forEach(lockItem -> lockClient.releaseLock(ReleaseLockOptions.builder(lockItem)
            .withBestEffort(true)
            .build())), executor);
  }
@Test
  void withLockAsync() throws InterruptedException {
    accountLockManager.withLockAsync(List.of(FIRST_NUMBER, SECOND_NUMBER),
        () -> CompletableFuture.completedFuture(null), executor).join();

    verify(lockClient, times(2)).acquireLock(any());
    verify(lockClient, times(2)).releaseLock(any(ReleaseLockOptions.class));
  }