Java 类java.util.concurrent.CompletionException 实例源码

/**
 * Verifies that inlining continuations do not have to complete execution before {@link AsyncAutoResetEvent#set()}
 * returns.
 */
@Test
public void testSetReturnsBeforeInlinedContinuations() throws Exception {
    CompletableFuture<Void> setReturned = new CompletableFuture<>();
    CompletableFuture<Void> inlinedContinuation = event.waitAsync().whenComplete((result, exception) -> {
        try {
            // Arrange to synchronously block the continuation until set() has returned,
            // which would deadlock if set() does not return until inlined continuations complete.
            setReturned.get(ASYNC_DELAY.toMillis(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException ex) {
            throw new CompletionException(ex);
        }
    });

    event.set();
    setReturned.complete(null);
    inlinedContinuation.get(ASYNC_DELAY.toMillis(), TimeUnit.MILLISECONDS);
}

@Test(timeout = 1000)
public void testAuthorizerThrowsRequiresHttpAction(final TestContext testContext) throws Exception {
    simulatePreviousAuthenticationSuccess();
    final AsyncClient<TestCredentials, TestProfile> indirectClient = getMockIndirectClient(NAME);
    final Clients<AsyncClient<? extends Credentials, ? extends CommonProfile>, AsyncAuthorizationGenerator<CommonProfile>> clients = new Clients<>(CALLBACK_URL, indirectClient);
    when(config.getClients()).thenReturn(clients);
    final String authorizers = NAME;
    addSingleAuthorizerToConfig((context, prof) -> { throw HttpAction.status("bad request", 400, context); });
    asyncSecurityLogic = new DefaultAsyncSecurityLogic<>(true, false, config, httpActionAdapter);
    final Async async = testContext.async();
    final CompletableFuture<Object> result = simulatePreviousAuthenticationSuccess()
            .thenCompose(v -> asyncSecurityLogic.perform(webContext, accessGrantedAdapter, null, authorizers, null));
    exception.expect(CompletionException.class);
    exception.expectCause(allOf(IsInstanceOf.instanceOf(HttpAction.class),
            hasProperty("message", is("bad request")),
            hasProperty("code", is(400))));
    assertSuccessfulEvaluation(result, ExceptionSoftener.softenConsumer(o -> {
        assertThat(o, is(nullValue()));
        assertThat(status.get(), is(400));
        verify(accessGrantedAdapter, times(0)).adapt(webContext);
    }), async);
}

private ArrayList<JsonNode> getResultListFromJson(JsonNode response) {

        ArrayList<JsonNode> resultList = new ArrayList<>();

        // ignore case when parsing items array
        String itemsKey = response.has("Items") ? "Items" : "items";

        for (JsonNode item : response.withArray(itemsKey)) {
            try {
                resultList.add(item);
            } catch (Exception e) {
                log.error("Could not parse data from Key Value Storage");
                throw new CompletionException(
                    new ExternalDependencyException(
                        "Could not parse data from Key Value Storage"));
            }
        }

        return resultList;
    }

@Test
public void testDoubleDirectClientChooseBadDirectClient(final TestContext testContext) throws Exception {
    final Clients<AsyncClient<? extends Credentials, ? extends CommonProfile>, AsyncAuthorizationGenerator<CommonProfile>> clients = doubleDirectClients();
    when(config.getClients()).thenReturn(clients);
    final String clientNames = NAME;
    when(webContext.getRequestParameter(eq(Clients.DEFAULT_CLIENT_NAME_PARAMETER))).thenReturn(VALUE);
    asyncSecurityLogic = new DefaultAsyncSecurityLogic<>(true, true, config, httpActionAdapter);
    final Async async = testContext.async();

    exception.expect(CompletionException.class);
    exception.expectCause(allOf(IsInstanceOf.instanceOf(TechnicalException.class),
            hasProperty("message", is("Client not allowed: " + VALUE))));
    final CompletableFuture<Object> result = asyncSecurityLogic.perform(webContext, accessGrantedAdapter, clientNames, null, null);
    assertSuccessfulEvaluation(result, ExceptionSoftener.softenConsumer(o -> {
        assertThat(o, is(nullValue()));
        verify(accessGrantedAdapter, times(0)).adapt(webContext);
    }), async);

}

@Test(timeout = 1000)
public void testAlreadyAuthenticatedNotAuthorized(final TestContext testContext) throws Exception {
    simulatePreviousAuthenticationSuccess();
    final AsyncClient<TestCredentials, TestProfile> indirectClient = getMockIndirectClient(NAME);
    final Clients<AsyncClient<? extends Credentials, ? extends CommonProfile>, AsyncAuthorizationGenerator<CommonProfile>> clients = new Clients<>(CALLBACK_URL, indirectClient);
    when(config.getClients()).thenReturn(clients);
    final String authorizers = NAME;
    addSingleAuthorizerToConfig((context, prof) -> prof.get(0).getId().equals(BAD_USERNAME));
    asyncSecurityLogic = new DefaultAsyncSecurityLogic<>(true, false, config, httpActionAdapter);
    final Async async = testContext.async();
    final CompletableFuture<Object> result = simulatePreviousAuthenticationSuccess()
            .thenCompose(v -> asyncSecurityLogic.perform(webContext, accessGrantedAdapter, null, authorizers, null));

    exception.expect(CompletionException.class);
    exception.expectCause(allOf(IsInstanceOf.instanceOf(HttpAction.class),
            hasProperty("message", is("forbidden")),
            hasProperty("code", is(403))));
    assertSuccessfulEvaluation(result, ExceptionSoftener.softenConsumer(o -> {
        assertThat(o, is(nullValue()));
        assertThat(status.get(), is(403));
        verify(accessGrantedAdapter, times(0)).adapt(webContext);
    }), async);
}

@Test
public void testRemoval() throws Exception {
    // A removal should fail when there's no such entry.
    assertThatThrownBy(() -> repo
            .commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, Change.ofRemoval(jsonPaths[0])).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(ChangeConflictException.class);

    Revision revision = repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, jsonUpserts[0]).join();
    assertThat(repo.exists(revision, jsonPaths[0]).join()).isTrue();

    // A removal should succeed when there's an entry.
    revision = repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, Change.ofRemoval(jsonPaths[0])).join();
    assertThat(repo.exists(revision, jsonPaths[0]).join()).isFalse();

    // A removal should fail when there's no such entry.
    assertThatThrownBy(() -> repo
            .commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, Change.ofRemoval(jsonPaths[0])).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(ChangeConflictException.class);
}

@Test
public void testRecursiveRemoval() throws Exception {
    // A recursive removal should fail when there's no such entry.
    final String curDir = prefix.substring(0, prefix.length() - 1); // Remove trailing '/'.
    assertThatThrownBy(
            () -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, Change.ofRemoval(curDir)).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(ChangeConflictException.class);

    // Add some files
    repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, jsonUpserts).join();
    assertThat(repo.find(HEAD, allPattern).join()).hasSize(jsonUpserts.length);

    // Perform a recursive removal
    repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, Change.ofRemoval(curDir)).join();
    assertThat(repo.find(HEAD, allPattern).join()).isEmpty();
}

@Test
public void testNoLockHeldForCancellationTokenContinuation() {
    CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
    CompletableFuture<GenericParameterHelper> pollFuture = queue.pollAsync(cancellationTokenSource.getToken());
    CompletableFuture<Void> handled = pollFuture.handle((result, exception) -> {
        try {
            ForkJoinPool.commonPool().submit(ExecutionContext.wrap(() -> {
                // Add presumably requires a private lock internally.
                // Since we're calling it on a different thread than the
                // blocking cancellation continuation, this should deadlock
                // if and only if the queue is holding a lock while invoking
                // our cancellation continuation (which they shouldn't be doing).
                queue.add(new GenericParameterHelper(1));
            })).get(ASYNC_DELAY.toMillis(), TimeUnit.MILLISECONDS);
            return null;
        } catch (InterruptedException | ExecutionException | TimeoutException ex) {
            throw new CompletionException(ex);
        }
    });

    cancellationTokenSource.cancel();

    handled.join();
    Assert.assertEquals(1, queue.size());
}

@Test
public void testRenameFailure() throws Exception {
    assertThatThrownBy(() -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY,
                                         jsonUpserts[0], jsonUpserts[1],
                                         Change.ofRename(jsonPaths[0], jsonPaths[1])).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(ChangeConflictException.class);

    // Renaming to its own path.
    repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, jsonUpserts[0]).join();
    assertThatThrownBy(() -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY,
                                         Change.ofRename(jsonPaths[0], jsonPaths[0])).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(ChangeConflictException.class);

    // Renaming to its own path, when the file is not committed yet.
    assertThatThrownBy(() -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY,
                                         jsonUpserts[1], Change.ofRename(jsonPaths[1], jsonPaths[1]))
                                 .join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(ChangeConflictException.class);
}

private static void connectAsync(String server) throws Exception {
    try {
        HttpClient client = HttpClient.newBuilder()
                .version(HttpClient.Version.HTTP_2)
                .build();
        HttpRequest request = HttpRequest.newBuilder(new URI(server))
                .timeout(Duration.ofMillis(TIMEOUT))
                .POST(fromString("body"))
                .build();
        HttpResponse<String> response = client.sendAsync(request, asString()).join();
        System.out.println("Received unexpected reply: " + response.statusCode());
        throw new RuntimeException("unexpected successful connection");
    } catch (CompletionException e) {
        if (e.getCause() instanceof HttpTimeoutException) {
            System.out.println("expected exception: " + e.getCause());
        } else {
            throw new RuntimeException("Unexpected exception received: " + e.getCause(), e);
        }
    }
}

@Test
public void testEmptyCommitWithRedundantUpsert() throws Exception {
    assertThatThrownBy(
            () -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, Collections.emptyList()).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(RedundantChangeException.class);

    // Create a file to produce redundant changes.
    repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, jsonUpserts[0]).join();

    // Ensure redundant changes do not count as a valid change.
    assertThatThrownBy(
            () -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, jsonUpserts[0]).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(RedundantChangeException.class);
}

@Test
public void testEmptyCommitWithRedundantUpsert2() throws Exception {
    // Create files to produce redundant changes.
    final Change<JsonNode> change1 = Change.ofJsonUpsert("/redundant_upsert_2.json",
                                                         "{ \"foo\": 0, \"bar\": 1 }");
    final Change<String> change2 = Change.ofTextUpsert("/redundant_upsert_2.txt", "foo");
    repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, change1).join();
    repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, change2).join();

    // Ensure redundant changes do not count as a valid change.
    assertThatThrownBy(() -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, change1).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(RedundantChangeException.class);
    assertThatThrownBy(() -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, change2).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(RedundantChangeException.class);

    // Ensure a change only whose serialized form is different does not count.
    final Change<JsonNode> change1a = Change.ofJsonUpsert("/redundant_upsert_2.json",
                                                          "{ \"bar\": 1, \"foo\": 0 }");
    assertThatThrownBy(() -> repo.commit(HEAD, 0L, Author.UNKNOWN, SUMMARY, change1a).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(RedundantChangeException.class);
}

protected <T> CompletableFuture<Void> assertSuccessfulEvaluation (final CompletableFuture<T> future,
                                                                  final Consumer<T> assertions,
                                                                  final Async async){
    return future.handle((v, t) -> {
        if (t != null) {
            executionContext.runOnContext(ExceptionSoftener.softenRunnable(() -> {
                if (t instanceof CompletionException) {throw t.getCause();
            }else {
                            throw t;
                        }
                    }));
        } else {
            executionContext.runOnContext(() -> {
                assertions.accept(v);
                async.complete();
            });
        }
        return null;
    });
}

@Test
public void testCreateProjectFailures() throws Exception {
    assertThatThrownByWithExpectedException(ProjectExistsException.class, "project: p", () ->
            rule.client().createProject(rule.project()).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(CentralDogmaException.class)
            .matches(e -> ((CentralDogmaException) e.getCause()).getErrorCode() == PROJECT_EXISTS);

    assertThatThrownByWithExpectedException(ProjectExistsException.class, "project: rp", () ->
            // It is not allowed to create a new project whose name is same with the removed project.
            rule.client().createProject(rule.removedProject()).join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(CentralDogmaException.class)
            .matches(e -> ((CentralDogmaException) e.getCause()).getErrorCode() == PROJECT_EXISTS);

    assertThatThrownByWithExpectedException(
            IllegalArgumentException.class, "invalid project name: ..", () ->
                    rule.client().createProject("..").join())
            .isInstanceOf(CompletionException.class).hasCauseInstanceOf(CentralDogmaException.class)
            .matches(e -> ((CentralDogmaException) e.getCause()).getErrorCode() == BAD_REQUEST);
}

@Test
public void testCreateRepositoryFailures() throws Exception {
    assertThatThrownByWithExpectedException(RepositoryExistsException.class, "repository: r", () ->
            rule.client().createRepository(rule.project(), rule.repo1()).join())
            .isInstanceOf(CompletionException.class).hasCauseInstanceOf(CentralDogmaException.class)
            .matches(e -> ((CentralDogmaException) e.getCause()).getErrorCode() == REPOSITORY_EXISTS);

    assertThatThrownByWithExpectedException(RepositoryExistsException.class, "repository: rr", () ->
            // It is not allowed to create a new repository whose name is same with the removed repository.
            rule.client().createRepository(rule.project(), rule.removedRepo()).join())
            .isInstanceOf(CompletionException.class).hasCauseInstanceOf(CentralDogmaException.class)
            .matches(e -> ((CentralDogmaException) e.getCause()).getErrorCode() == REPOSITORY_EXISTS);

    assertThatThrownByWithExpectedException(
            IllegalArgumentException.class, "invalid repository name: ..", () ->
                    rule.client().createRepository(rule.project(), "..").join())
            .isInstanceOf(CompletionException.class).hasCauseInstanceOf(CentralDogmaException.class)
            .matches(e -> ((CentralDogmaException) e.getCause()).getErrorCode() == BAD_REQUEST);
}

/**
 * Reports result using Future.get conventions.
 */
private static <T> T reportGet(Object r)
    throws InterruptedException, ExecutionException {
    if (r == null) // by convention below, null means interrupted
        throw new InterruptedException();
    if (r instanceof AltResult) {
        Throwable x, cause;
        if ((x = ((AltResult)r).ex) == null)
            return null;
        if (x instanceof CancellationException)
            throw (CancellationException)x;
        if ((x instanceof CompletionException) &&
            (cause = x.getCause()) != null)
            x = cause;
        throw new ExecutionException(x);
    }
    @SuppressWarnings("unchecked") T t = (T) r;
    return t;
}

@Override
public void render(final PebbleTemplateImpl self, Writer writer, final EvaluationContextImpl context)
        throws PebbleException, IOException {
    try {
        final String body;
        Cache tagCache = context.getTagCache();
        if(isNull(tagCache)){ //No cache
            body = render(self, context);
        }
        else {
            CacheKey key = new CacheKey((String) this.name.evaluate(self, context), context.getLocale());
            body = (String) context.getTagCache().get(key, k -> {
                try {
                    return render(self, context);
                } catch (PebbleException e) {
                    throw new RuntimePebbleException(e);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            });
        }
        writer.write(body);
    } catch (CompletionException e) {
        throw new PebbleException(e, "Could not render cache block [" + this.name + "]");
    }
}

@Test
public void testDirectClientThrowsRequiresHttpAction(final TestContext testContext) throws Exception {
    final AsyncClient directClient = getMockDirectClient(NAME, TEST_CREDENTIALS);
    final Clients<AsyncClient<? extends Credentials, ? extends CommonProfile>, AsyncAuthorizationGenerator<CommonProfile>> clients = new Clients<>(CALLBACK_URL, directClient);
    when(config.getClients()).thenReturn(clients);
    final String clientNames = NAME;
    asyncSecurityLogic = new DefaultAsyncSecurityLogic<>(true, false, config, httpActionAdapter);
    when(directClient.getCredentials(eq(webContext))).thenReturn(delayedException(250,
            (() -> HttpAction.status("bad request", 400, webContext))));
    final Async async = testContext.async();
    final CompletableFuture<Object> result = asyncSecurityLogic.perform(webContext, accessGrantedAdapter, clientNames, null, null);
    exception.expect(CompletionException.class);
    exception.expectCause(allOf(IsInstanceOf.instanceOf(HttpAction.class),
            hasProperty("message", is("bad request")),
            hasProperty("code", is(400))));

    assertSuccessfulEvaluation(result, ExceptionSoftener.softenConsumer(o -> {
        assertThat(o, is(nullValue()));
        assertThat(status.get(), is(400));
        verify(accessGrantedAdapter, times(0)).adapt(webContext);
    }), async);
}

/**
 * Verifies that continuations scheduled on a future will not be executed inline with the specified completing
 * action.
 *
 * @param antecedent The future to test.
 * @param completingAction The action that results in the synchronous completion of the future.
 */
protected static void verifyDoesNotInlineContinuations(@NotNull CompletableFuture<?> antecedent, @NotNull Runnable completingAction) {
    Requires.notNull(antecedent, "antecedent");
    Requires.notNull(completingAction, "completingAction");

    CompletableFuture<Void> completingActionFinished = new CompletableFuture<>();
    CompletableFuture<Void> continuation = antecedent.handle((result, exception) -> {
        try {
            return completingActionFinished.get(ASYNC_DELAY.toMillis(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException ex) {
            throw new CompletionException(ex);
        }
    });

    completingAction.run();
    completingActionFinished.complete(null);

    // Rethrow the exception if it turned out it deadlocked.
    continuation.join();
}

@Override
public boolean completeExceptionally(Throwable ex) {
    if (ex instanceof CompletionException) {
        // if the exception is wrapped inside completion exception, it did not happen immediately in previous
        // stage, as described in CompletionStage documentation:
        // " if a stage's computation terminates abruptly with an (unchecked) exception or error,
        // then all dependent stages requiring its completion complete exceptionally as well,
        // with a CompletionException holding the exception as its cause."
        // notice how it doesn't say that this doesn't hold for first dependent, because that doesn't yet form
        // a dependent stage, only the result of the callback forms a dependent stage.
        // There's a discussion when this was closed as not a bug
        // http://bugs.java.com/view_bug.do?bug_id=8068432
        //
        super.completeExceptionally(ex);
    } else {
        compensationLogic.apply(ex).whenComplete((r,t) -> {
                if (t == null) {
                    complete(r);
                } else {
                    super.completeExceptionally(t);
                }
            });
        }
    return false;
}

/**
 * Returns a new future instance who's lifecycle depends on this instance but not vice versa.
 *
 * <p>
 * If this instance gets completed (either normally or with an error) then forked instance will be completed too. Completing forked
 * instance has no impact on this instance.
 * </p>
 *
 * @return Fork of this instance.
 */
public F fork() {
    F dep = newInstance();

    if (isSuccess()) {
        dep.complete(getNow(null));
    } else if (isCancelled()) {
        dep.cancel(false);
    } else {
        whenComplete((result, error) -> {
            if (isCancelled()) {
                dep.cancel(false);
            } else if (error != null) {
                if (error instanceof CompletionException && error.getCause() != null) {
                    dep.completeExceptionally(error.getCause());
                } else {
                    dep.completeExceptionally(error);
                }
            } else {
                dep.complete(result);
            }
        });
    }

    return dep;
}

@Test
public void resolveCustomTypeReference_WithExceptionOnCustomTypeFetch_ShouldNotResolveReferences() {
    final InventoryEntryDraftBuilder draftBuilder = InventoryEntryDraftBuilder
        .of(SKU, QUANTITY, DATE_1, RESTOCKABLE_IN_DAYS, Channel.referenceOfId(UUID_KEY))
        .custom(CustomFieldsDraft.ofTypeIdAndJson(CUSTOM_TYPE_KEY, new HashMap<>()));

    when(typeService.fetchCachedTypeId(anyString()))
        .thenReturn(CompletableFutureUtils.failed(new SphereException("bad request")));

    final InventoryReferenceResolver referenceResolver =
        new InventoryReferenceResolver(syncOptions, typeService, channelService);

    referenceResolver.resolveCustomTypeReference(draftBuilder)
                     .exceptionally(exception -> {
                         assertThat(exception).isExactlyInstanceOf(CompletionException.class);
                         assertThat(exception.getCause()).isExactlyInstanceOf(SphereException.class);
                         assertThat(exception.getCause().getMessage()).contains("bad request");
                         return null;
                     }).toCompletableFuture().join();
}

@Test
public void resolveCustomTypeReference_WithKeyAsUuidSetAndNotAllowed_ShouldNotResolveCustomTypeReference() {
    final InventoryEntryDraftBuilder draftBuilder = InventoryEntryDraftBuilder
        .of(SKU, QUANTITY, DATE_1, RESTOCKABLE_IN_DAYS, Channel.referenceOfId(CHANNEL_KEY))
        .custom(CustomFieldsDraft.ofTypeIdAndJson(UUID_KEY, new HashMap<>()));

    final InventoryReferenceResolver referenceResolver =
        new InventoryReferenceResolver(syncOptions, typeService, channelService);

    referenceResolver.resolveCustomTypeReference(draftBuilder)
                     .exceptionally(exception -> {
                         assertThat(exception).isExactlyInstanceOf(CompletionException.class);
                         assertThat(exception.getCause())
                             .isExactlyInstanceOf(ReferenceResolutionException.class);
                         assertThat(exception.getCause().getMessage())
                             .isEqualTo("Failed to resolve custom type reference on InventoryEntryDraft"
                                     + " with SKU:'1000'. Reason: Found a UUID in the id field. Expecting a key"
                                     + " without a UUID value. If you want to allow UUID values for reference keys,"
                                     + " please use the allowUuidKeys(true) option in the sync options.");
                         return null;
                     }).toCompletableFuture().join();
}

@Test
public void resolveCustomTypeReference_WithEmptyIdOnCustomTypeReference_ShouldNotResolveCustomTypeReference() {
    final InventoryEntryDraftBuilder draftBuilder = InventoryEntryDraftBuilder
        .of(SKU, QUANTITY, DATE_1, RESTOCKABLE_IN_DAYS, Channel.referenceOfId(CHANNEL_KEY))
        .custom(CustomFieldsDraft.ofTypeIdAndJson("", new HashMap<>()));

    final InventoryReferenceResolver referenceResolver =
        new InventoryReferenceResolver(syncOptions, typeService, channelService);

    referenceResolver.resolveCustomTypeReference(draftBuilder)
                     .exceptionally(exception -> {
                         assertThat(exception).isExactlyInstanceOf(CompletionException.class);
                         assertThat(exception.getCause())
                             .isExactlyInstanceOf(ReferenceResolutionException.class);
                         assertThat(exception.getCause().getMessage())
                             .isEqualTo("Failed to resolve custom type reference on InventoryEntryDraft"
                                     + " with SKU:'1000'. Reason: Reference 'id' field value is blank"
                                     + " (null/empty).");
                         return null;
                     }).toCompletableFuture().join();
}

/**
 * Verifies that inlining continuations do not have to complete execution before {@link AsyncManualResetEvent#set()}
 * returns.
 */
@Test
public void testSetReturnsBeforeInlinedContinuations() throws Exception {
    CompletableFuture<Void> setReturned = new CompletableFuture<>();
    CompletableFuture<Void> inlinedContinuation = event.waitAsync()
        .thenRun(()
            -> {
            try {
                // Arrange to synchronously block the continuation until set() has returned,
                // which would deadlock if set() does not return until inlined continuations complete.
                setReturned.get(ASYNC_DELAY.toMillis(), TimeUnit.MILLISECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException ex) {
                throw new CompletionException(ex);
            }
        });
    event.set();
    Assert.assertTrue(event.isSet());
    setReturned.complete(null);
    inlinedContinuation.get(ASYNC_DELAY.toMillis(), TimeUnit.MILLISECONDS);
}

@Test
public void sync_WithExistingCategoryButWithEmptyCustomTypeReference_ShouldFailSync() {
    final Category mockCategory = getMockCategory(UUID.randomUUID().toString(), "key");
    final CategoryService mockCategoryService =
        mockCategoryService(singleton(mockCategory), emptySet(), mockCategory);
    final CategorySync categorySync = new CategorySync(categorySyncOptions, getMockTypeService(),
        mockCategoryService);
    final List<CategoryDraft> categoryDrafts = singletonList(
        getMockCategoryDraft(Locale.ENGLISH, "name", "key", "parentKey",
            "", new HashMap<>()));

    final CategorySyncStatistics syncStatistics = categorySync.sync(categoryDrafts).toCompletableFuture().join();

    assertThat(syncStatistics).hasValues(1, 0, 0, 1);
    assertThat(errorCallBackMessages).hasSize(1);
    assertThat(errorCallBackMessages.get(0)).isEqualTo(format("Failed to resolve references on CategoryDraft with"
        + " key:'key'. Reason: %s: Failed to resolve custom type reference on "
        + "CategoryDraft with key:'key'. Reason: Reference 'id' field value is blank (null/"
        + "empty).", ReferenceResolutionException.class.getCanonicalName()));
    assertThat(errorCallBackExceptions).hasSize(1);
    assertThat(errorCallBackExceptions.get(0)).isExactlyInstanceOf(CompletionException.class);
    assertThat(errorCallBackExceptions.get(0).getCause()).isExactlyInstanceOf(ReferenceResolutionException.class);
}

@Test
public void sync_WithNotAllowedUuidCustomTypeKey_ShouldFailSync() {
    final Category mockCategory = getMockCategory(UUID.randomUUID().toString(), "key");
    final CategoryService mockCategoryService =
        mockCategoryService(singleton(mockCategory), emptySet(), mockCategory);
    final CategorySync categorySync = new CategorySync(categorySyncOptions, getMockTypeService(),
        mockCategoryService);
    final List<CategoryDraft> categoryDrafts = singletonList(
        getMockCategoryDraft(Locale.ENGLISH, "name", "key", "parentKey",
            UUID.randomUUID().toString(), new HashMap<>()));

    final CategorySyncStatistics syncStatistics = categorySync.sync(categoryDrafts).toCompletableFuture().join();

    assertThat(syncStatistics).hasValues(1, 0, 0, 1);
    assertThat(errorCallBackMessages).hasSize(1);
    assertThat(errorCallBackMessages.get(0)).isEqualTo(format("Failed to resolve references on CategoryDraft with"
        + " key:'key'. Reason: %s: Failed to resolve custom type reference on "
        + "CategoryDraft with key:'key'. Reason: Found a UUID in the id field. Expecting a key"
        + " without a UUID value. If you want to allow UUID values for reference keys, please use the"
        + " allowUuidKeys(true) option in the sync options.",
        ReferenceResolutionException.class.getCanonicalName()));
    assertThat(errorCallBackExceptions).hasSize(1);
    assertThat(errorCallBackExceptions.get(0)).isExactlyInstanceOf(CompletionException.class);
    assertThat(errorCallBackExceptions.get(0).getCause()).isExactlyInstanceOf(ReferenceResolutionException.class);
}

/**
 * Reports result using Future.get conventions.
 */
private static <T> T reportGet(Object r)
    throws InterruptedException, ExecutionException {
    if (r == null) // by convention below, null means interrupted
        throw new InterruptedException();
    if (r instanceof AltResult) {
        Throwable x, cause;
        if ((x = ((AltResult)r).ex) == null)
            return null;
        if (x instanceof CancellationException)
            throw (CancellationException)x;
        if ((x instanceof CompletionException) &&
            (cause = x.getCause()) != null)
            x = cause;
        throw new ExecutionException(x);
    }
    @SuppressWarnings("unchecked") T t = (T) r;
    return t;
}

@Test
@Ignore
public void timeoutTestIgn() throws Exception {
    CoapServer cnn = CoapServerBuilder.newBuilder().transport(61616, Executors.newCachedThreadPool()).build();
    cnn.start();

    CoapPacket request = new CoapPacket(new InetSocketAddress(InetAddress.getLocalHost(), 60666));
    request.setMethod(Method.GET);
    request.headers().setUriPath("/test/1");
    request.setMessageId(1647);

    try {
        cnn.makeRequest(request).join();
        fail("Exception was expected");
    } catch (CompletionException ex) {
        //expected
    }
    assertEquals("Wrong number of transactions", 0, ((CoapUdpMessaging) cnn.getCoapMessaging()).getNumberOfTransactions());
    cnn.stop();

}

static void test10(String s) throws Exception {
    System.out.print("test10: " + s);
    URI uri = new URI(s);
    RedirectErrorHandler handler = uri.getScheme().equals("https")
            ? redirectErrorHandlerSecure : redirectErrorHandler;

    HttpRequest request = HttpRequest.newBuilder(uri).GET().build();
    CompletableFuture<HttpResponse<String>> cf =
            client.sendAsync(request, asString());

    try {
        HttpResponse<String> response = cf.join();
        throw new RuntimeException("Exepected Completion Exception");
    } catch (CompletionException e) {
        //System.out.println(e);
    }

    System.out.printf(" (Calls %d) ", handler.count());
    System.out.println(" OK");
}

@Test
public void testJoinCancellation() {
    // Kick off the BeginAsync work from a background thread that has no special
    // affinity to the main thread.
    JoinableFuture<Void> joinable = asyncPump.runAsync(() -> {
        return Async.awaitAsync(
            Async.yieldAsync(),
            () -> Async.awaitAsync(
                asyncPump.switchToMainThreadAsync(),
                () -> Async.awaitAsync(Async.delayAsync(ASYNC_DELAY))));
    });

    Assert.assertFalse(joinable.getFuture().isDone());
    CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(ASYNC_DELAY.dividedBy(4));

    // This isn't the cancellation behavior we want...
    thrown.expect(CompletionException.class);
    thrown.expectCause(isA(CancellationException.class));
    joinable.join(cancellationTokenSource.getToken());
}

@Test
public void testDoubleIndirectClientBadOneChosen(final TestContext testContext) throws Exception {
    final AsyncClient<TestCredentials, TestProfile> indirectClient = getMockIndirectClient(NAME, PAC4J_URL);
    final AsyncClient<TestCredentials, TestProfile> indirectClient2 = getMockIndirectClient(VALUE, PAC4J_BASE_URL);

    final Clients<AsyncClient<? extends Credentials, ? extends CommonProfile>, AsyncAuthorizationGenerator<CommonProfile>> clients = new Clients<>(CALLBACK_URL, indirectClient, indirectClient2);

    when(config.getClients()).thenReturn(clients);
    final String clientNames = NAME;
    when(webContext.getRequestParameter(eq(Clients.DEFAULT_CLIENT_NAME_PARAMETER))).thenReturn(VALUE);

    asyncSecurityLogic = new DefaultAsyncSecurityLogic<>(true, false, config, httpActionAdapter);

    final Async async = testContext.async();

    exception.expect(CompletionException.class);
    exception.expectCause(allOf(IsInstanceOf.instanceOf(TechnicalException.class),
            hasProperty("message", is("Client not allowed: " + VALUE))));
    final CompletableFuture<Object> result = asyncSecurityLogic.perform(webContext, accessGrantedAdapter, clientNames, null, null);
    assertSuccessfulEvaluation(result, ExceptionSoftener.softenConsumer(o -> {
        assertThat(o, is(nullValue()));
        verify(accessGrantedAdapter, times(0)).adapt(webContext);
    }), async);
}

protected final void simulateUIThread(Supplier<CompletableFuture<?>> testMethod) {
    Verify.operation(this.originalThread == Thread.currentThread(), "We can only simulate the UI thread if you're already on it (the starting thread for the test).");

    StrongBox<Throwable> failure = new StrongBox<>(null);
    this.dispatcherContext.post(
        state -> {
            Async.finallyAsync(
                Async.awaitAsync(testMethod.get())
                .exceptionally(ex -> {
                    failure.value = ex;
                    return null;
                }),
                () -> testFrame.setContinue(false));
        },
        null);

    SingleThreadedSynchronizationContext.pushFrame(this.dispatcherContext, this.testFrame);
    if (failure.value != null) {
        // Rethrow original exception without rewriting the callstack.
        throw new CompletionException(failure.value);
    }
}

private ServiceInstance createInstance(TestDescriptor descriptor, DockerClient client, int i) {
      CreateContainerCmd cmd = client.createContainerCmd(descriptor.getImage().value());
      cmd = computeContainerName(descriptor, i, cmd);
      cmd = executeOptionBuilders(descriptor, cmd);
      if (descriptor.getCustomisationHook() != null) {
          cmd = executeCustomisationHook(descriptor.getCustomisationHook(), descriptor.getInstance(), cmd);
      }
      String containerId = null;
      Status status = null;
      String statusDetails = null;
try {
    containerId = createAndStartContainer(cmd, descriptor.getImage().pull(),  client);
    status = Status.STARTED;
    statusDetails = "Started.";
} catch (Throwable t) {
    if(t instanceof CompletionException) {
        if(t.getCause() != null && t.getCause() instanceof ContainerException) {
            containerId = ((ContainerException) t.getCause()).getContainerId();
            statusDetails = t.getCause().getCause() != null ? t.getCause().getCause().getMessage() : null;
        } else {
            statusDetails = t.getCause() != null ? t.getCause().getMessage() : null;
        }   
    } else {
        statusDetails = t.getMessage();
    }
    status = Status.ABORTED;
}
return ServiceInstance.builder()
              .containerName(cmd.getName())
              .containerId(containerId)
              .status(status)
              .statusDetails(statusDetails)
              .build();
  }

/**
 * Makes a GET request and returns the result
 * 
 * @param url
 * @return HttpResult
 */
public CompletableFuture<HttpResult> get(String url) {
    return CompletableFuture.supplyAsync(() -> {
        try {
            return getSynch(new URL(url));
        } catch (Exception e) {
            throw new CompletionException(e);
        }
    });
}

/**
 * Makes a GET request synchronously
 * 
 * @param url
 * @return HttpResult
 */
private HttpResult getSynch(URL url) {
    try {
        HttpURLConnection conn = (HttpURLConnection) url.openConnection();
        conn.setRequestMethod("GET");
        HttpResult result = new HttpResult(IOUtils.toByteArray(conn.getInputStream()));
        return result;
    } catch (Exception e) {
        throw new CompletionException(e);
    }
}

@Test
public void testGetSynchronizationContext_DifferentThread() {
    Thread currentThread = Thread.currentThread();
    CompletableFuture<Void> asyncTest = Futures.runAsync(() -> {
        ThreadProperties.getSynchronizationContext(currentThread);
    });

    thrown.expect(CompletionException.class);
    thrown.expectCause(isA(AssertionError.class));
    asyncTest.join();
}

private static <E extends Throwable> Optional<E> unwrapFutureException(
    Class<E> causeType, Throwable exception) {
  for (Throwable e = exception; ; e = e.getCause()) {
    if (causeType.isInstance(e)) {
      return Optional.of(causeType.cast(e));
    }
    if (!(e instanceof ExecutionException || e instanceof CompletionException)) {
      return Optional.empty();
    }
  }
}

/**
 * Verifies that the exception is returned in a future rather than thrown from the asynchronous method.
 */
@Test
public void testSignalAndWaitAsyncReturnsFailedFutureOnError() {
    AsyncCountdownEvent event = new AsyncCountdownEvent(0);
    CompletableFuture<Void> result = event.signalAndWaitAsync();
    Assert.assertTrue(result.isCompletedExceptionally());
    Assert.assertFalse(result.isCancelled());

    thrown.expect(CompletionException.class);
    thrown.expectCause(isA(IllegalStateException.class));
    result.join();
}

@Test
public void testNoLockHeldForCancellationContinuation() {
    CompletableFuture<GenericParameterHelper> pollFuture = queue.pollAsync();
    CompletableFuture<Void> handled = pollFuture.handle((result, exception) -> {
        try {
            ForkJoinPool.commonPool().submit(ExecutionContext.wrap(() -> {
                // Add presumably requires a private lock internally.
                // Since we're calling it on a different thread than the
                // blocking cancellation continuation, this should deadlock
                // if and only if the queue is holding a lock while invoking
                // our cancellation continuation (which they shouldn't be doing).
                queue.add(new GenericParameterHelper(1));
            })).get(ASYNC_DELAY.toMillis(), TimeUnit.MILLISECONDS);
            return null;
        } catch (InterruptedException | ExecutionException | TimeoutException ex) {
            throw new CompletionException(ex);
        }
    });

    pollFuture.cancel(true);

    handled.join();
    Assert.assertEquals(1, queue.size());
}