0

I'm adding some code to an existing endpoint to send an email. We don't need the result of sending an email to return a response to the user, so I'm adding a .whenComplete() at the end of the chain of futures, calling our email service from within. The call to the email service is also async, returning a CompletionStage<Void>.

CompletionStage<SomeResponse> someEndpoint() {
  return doThings()
      .thenApply(things -> {
        return someResponseFormat(things);
      })
      .whenComplete((someResponse, ex) -> {
        if (ex == null) {
          emailClient.sendEmail(someResponse); // CompletionStage<Void>
        }
      });
}

As I understand, that task will be scheduled and executed. Do I need to call join() on sendEmail(...)? Would doing so have a different behavior than not calling them? What is the best practice?

Edit: Originally I asked if I need to call join() or get(), which was misunderstood as "which do I need to call," when I meant, "do I need to call either at all."

Andrew Cheong
  • 29,362
  • 15
  • 90
  • 145
  • When you call `#whenComplete`, it will execute the following task/lambda on the same thread as the previous `CompletionStage`. Thus, if `doThings()` is performing work on a thread pool, then the `whenComplete` will also perform on the same thread in the thread pool. If you wish to consume a thread in a different thread pool / executor / et al, then you can use `#whenCompleteAsync`. To answer the direct question, there's no need for `#join` or `#get` to be called, unless you want to call those outside of the `CompletableFuture` in order to wait on the task in the primary thread. – Rogue Jun 17 '22 at 13:29
  • I see. That I understand. I guess my question is more that I'm used to always seeing a "terminator" to futures—one of the thenApply / thenCompose / thenRun / etc.—or a join() or get() (mainly in tests)—so my fear is that not having a "terminator" may cause that code not to execute at all or execute at a random time that could take a while. – Andrew Cheong Jun 17 '22 at 13:32
  • 1
    Indeed, `java.util.Stream` works in this fashion. However, `CompletableFuture` will execute regardless of a dependent stage such as `whenComplete` or `thenRun`. The static utility methods will automatically begin the future, the only time they won't automatically run is if you construct the object manually. – Rogue Jun 17 '22 at 13:33
  • Ah! I didn't even realize that Java streams are the reason I was thinking that. Thanks for understanding and clearing it up. If you'd like to add an answer (namely that this is unlike streams), I think others with the same confusion may find the answer useful, and I'd be happy to accept. – Andrew Cheong Jun 17 '22 at 13:36
  • 1
    I think @Holger's answer meets the needs of the question fine, I've never really ground for rep ;) – Rogue Jun 17 '22 at 13:38
  • @Rogue - Okay, whoa. So I committed the code without a join() or get(). Tests I wrote to check that the .whenComplete() code ran, failed. This reminded me: I've seen this before—non-"terminated" futures for some reason work for me running locally (IntelliJ, docker)—but in CI/CD (Jenkins) fail consistently. I added a join() and now tests are passing. I've seen this several times before too—I think that's why I was thinking this, not just because of Java streams. Why does this happen? – Andrew Cheong Jun 17 '22 at 15:02
  • 1
    As a guess, it may be something within Jenkins itself which watches for the end of the main thread's execution, and then shuts down thread pools like the `ForkJoinPool#commonPool`. In this case, you would need the `#join` to keep the service running, whereas in a "baremetal" environment the pool will continue to run until the threads are reclaimed. You would have to test the potential circumstances. – Rogue Jun 17 '22 at 16:00

1 Answers1

3

The operation associated with emailClient.sendEmail(someResponse) will be scheduled regardless of whether you wait for its completion, so unless the JVM terminates in the meanwhile, it will complete. But

  • Nobody will notice when the operation completed or be able to wait for its completion.

  • Nobody will notice when the operation fails with an exception.

So what you probably want to do, is

CompletionStage<SomeResponse> someEndpoint() {
  return doThings()
      .thenApply(things -> someResponseFormat(things))
      .thenCompose(someResponse -> emailClient.sendEmail(someResponse)
          .thenApply(_void -> someResponse));
}

Then, when the caller of someEndpoint() invokes join() on it, the join() would wait for the completion of the sendEmail and also report errors when sendEmail fails. Likewise, when the caller of someEndpoint() chains dependent operations, they would start after the completion of sendEmail.

Holger
  • 285,553
  • 42
  • 434
  • 765
  • Thank you for your detailed answers (you've helped me quite a lot in other questions)! In this case, yup, I have the same understanding. But specifically I wanted _not_ to wait for completion, because of how slow our email service can be to respond. And that's why I came across my question—I've never made an async call from within a .whenComplete() before, and didn't know if there was something I should be wary about. – Andrew Cheong Jun 17 '22 at 13:39
  • 1
    Well, this version doesn’t wait for the completion, but leaves it to the caller to decide whether to wait for the completion. A reasonable strategy for the caller would be to do something like `someEndpoint().whenComplete((someResponse, ex) -> { if(ex != null) { log(ERROR, …, ex); } else { log(INFO, "completed with " + someResponse); } });` which still doesn’t wait for the completion but will eventually report about it. But as said, for your original version, the operation will be scheduled, regardless of whether you wait for its completion. – Holger Jun 17 '22 at 13:45
  • Oh I see now. You're doing a .thenApply after the .thenCompose. I see, I didn't know you could do that—that's good to know. Not related to your answer but, I've just run into some weirdness I've run into before: Without "terminating" a future, for some reason things work as expected when I run locally, but fail in Jenkins. (More on this in comments to original question ^.) – Andrew Cheong Jun 17 '22 at 15:05
  • 2
    That’s why I wrote “*unless the JVM terminates in the meanwhile*”, which can become a problem, depending on the environment. Especially with testcases, you need a construct to prevent the test environment from terminating the test as “failed” after some time. If you don’t find a suitable built-in control mechanism, `ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.DAYS);` at the end of a test case, to ensure that there are no pending async operations, may help. – Holger Jun 17 '22 at 15:13