@Override
  public <U> ParSeqBasedCompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,
      Executor executor)
  {
    return nextStageByComposingTask(_task.flatMap("thenCompose", t -> Task.async(() -> {
      final SettablePromise<U> promise = Promises.settable();
      executor.execute(() -> {
        CompletionStage<? extends U> future = fn.apply(t);
        future.whenComplete((value, exception) -> {
          if (exception != null) {
            promise.fail(exception);
          } else {
            promise.done(value);
          }
        });
      });
      return promise;
    })));
  }
@Test
  public void testThenComposeAsync() throws Exception
  {
    Consumer<String> consumer = mock(Consumer.class);
    CountDownLatch waitLatch = new CountDownLatch(1);

    CompletionStage<String> completionStage = createTestStage(TESTVALUE1);
    CompletionStage<String> completionStage2 = createTestStage(TESTVALUE2);
    finish(completionStage.thenComposeAsync(r -> {
      assertEquals(THREAD_NAME_VALUE, Thread.currentThread().getName());
      waitLatch.countDown();
      return completionStage2;
    }, _mockExecutor).thenAccept(consumer));

    waitLatch.await(1000, TimeUnit.MILLISECONDS);
    verify(consumer, times(1)).accept(TESTVALUE2);
  }