public void withLock(final List<String> e164s, final Runnable task, final Executor lockAcquisitionExecutor) {
    if (e164s.isEmpty()) {
      throw new IllegalArgumentException("List of e164s to lock must not be empty");
    }

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

    try {
      // Offload the acquire/release tasks to the dedicated lock acquisition executor. The lock client performs blocking
      // operations while holding locks which forces thread pinning when this method runs on a virtual thread.
      // https://github.com/awslabs/amazon-dynamodb-lock-client/issues/97
      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);
          }
        }
      }, lockAcquisitionExecutor).join();

      task.run();
    } finally {
      CompletableFuture.runAsync(() -> {
        for (final LockItem lockItem : lockItems) {
          lockClient.releaseLock(ReleaseLockOptions.builder(lockItem)
              .withBestEffort(true)
              .build());
        }
      }, lockAcquisitionExecutor).join();
    }
  }
@Test
  void withLockTaskThrowsException() throws InterruptedException {
    assertThrows(RuntimeException.class, () -> accountLockManager.withLock(List.of(FIRST_NUMBER, SECOND_NUMBER), () -> {
      throw new RuntimeException();
    }, executor));

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