MicroProfile Concurrency introduces APIs for propagating contexts across units of work that are thread-agnostic. It makes it possible to propagate context that was traditionally associated to the current thread across various types of units of work such as CompletionStage, CompletableFuture, Function, Runnable regardless of which particular thread ends up executing them.
When using a reactive model, execution is often cut into small units of work that are chained together to assemble a reactive pipeline. The context under which each unit of work execute is often unpredictable and depends on the particular reactive toolkit used. Some units might run with the context of a thread that awaits completion, or the context of a previous unit that completed and triggered the dependent unit, or with no/undefined context at all. Existing solutions for transferring thread context, such as the EE Concurrency Utilities ContextService, are tied to a specific reactive model, promotes usage of thread pools, is difficult to use and require a lot of boilerplate code. This specification makes it possible for thread context propagation to easily be done in a type-safe way, keeping boilerplate code to a minimum, as well as allowing for thread context propagation to be done automatically for many types of reactive models.
We distinguish two main use-cases for propagating contexts to reactive pipelines:
-
Splitting units of work into a sequential pipeline where each unit will be executed after the other. Turning an existing blocking request into an async request would produce such pipelines.
-
Fanning out units of work to be executed in parallel on a managed thread pool. Launching an asynchronous job from a request without waiting for its termination would produce such pipelines.
Goals
-
Pluggable context propagation to the most common unit of work types.
-
Mechanism for thread context propagation to
CompletableFutureandCompletionStageunits of work that reduces the need for boilerplate code. -
Full compatibility with EE Concurrency spec, such that proposed interfaces can seamlessly work alongside EE Concurrency, without depending on it.
This specification introduces two interfaces that contain methods that can work alongside Jakarta EE Concurrency, if available.
The interface, org.eclipse.microprofile.concurrent.ManagedExecutor, provides methods for obtaining managed instances of CompletableFuture which are backed by the managed executor as the default asynchronous execution facility and the default mechanism of defining thread context propagation. Similar to EE Concurrency’s ManagedExecutorService, the MicroProfile ManagedExecutor also implements the Java SE java.util.concurrent.ExecutorService interface, using managed threads when asynchronous invocation is required. It is possible for a single implementation to be capable of simultaneously implementing both ManagedExecutor and ManagedExecutorService interfaces.
A second interface, org.eclipse.microprofile.concurrent.ThreadContext, provides methods for individually contextualizing units of work such as CompletionStage, CompletionFuture, Runnable, Function, Supplier and more, without tying them to a particular thread execution model. This gives the user finer-grained control over the capture and propagation of thread context by remaining thread execution agnostic. It is possible for a single implementation to be capable of simultaneously implementing both ThreadContext and ContextService interfaces.
Instances of ManagedExecutor and ThreadContext can be constructed via builders with fluent API, for example,
ManagedExecutor executor = ManagedExecutor.builder()
.propagated(ThreadContext.APPLICATION)
.maxAsync(5)
.build();
ThreadContext threadContext = ThreadContext.builder()
.propagated(ThreadContext.APPLICATION, ThreadContext.CDI)
.cleared(ThreadContext.ALL_REMAINING)
.build();Applications should shut down instances of ManagedExecutor that they build after they are no longer needed. The shutdown request serves as a signal notifying the container that resources can be safely cleaned up.
Instances of ManagedExecutor and ThreadContext can be injected into CDI beans via the @Inject annotation.
Injection points with the @NamedInstance qualifier have a name that is explicitly declared by the qualifier.
Injection points that have no CDI qualifier on them have a unique name which is deduced from the injection point itself.
Container provides, per unique name, an instance of either ManagedExecutor and/or ThreadContext unless the application specifies a CDI producer with @NamedInstance qualifier, given unique name and respective type.
for example:
@Inject ManagedExecutor executor;
@Inject ThreadContext threadContext;
...
CompletableFuture<Integer> stage = executor
.supplyAsync(supplier1)
.thenApplyAsync(function1)
.thenApply(function2);
...
unmanagedCompletionStage.thenApply(threadContext.contextualFunction(function3));In the absence of other qualifiers and annotations, the container creates and injects a new default instance of ManagedExecutor or ThreadContext per injection point, as shown above.
The configuration of these instances is equivalent to the default values of @ManagedExecutorConfig and @ThreadContextConfig.
The @ManagedExecutorConfig and @ThreadContextConfig annotations provide the ability to configure the injection points with desired capabilities.
With just the config annotation in place, every injection point is assigned a new instance.
@Inject
@ManagedExecutorConfig(maxAsync = 5, propagated = ThreadContext.SECURITY)
ManagedExecutor configuredExecutor;
...
CompletableFuture<Long> stage = configuredExecutor
.newIncompleteFuture()
.thenApply(function)
.thenAccept(consumer);
stage.completeAsync(supplier);Configured instances of ManagedExecutor and ThreadContext are shared based on their unique name, which is assigned by the CDI qualifier, @NamedInstance.
When one injection point is annotated with both a configuration and a @NamedInstance qualifier, the application can inject the same instance elsewhere by annotating other injection points with just the matching qualifier.
In the example below, the executor1 injection point defines a configuration and a name myExec for a configured ManagerExecutor instance.
The other injection points, executor2 and executor3, share the same ManagedExecutor instance by using the @NamedInstance qualifier with the name myExec.
@Inject
@NamedInstance("myExec")
@ManagedExecutorConfig(propagated = { ThreadContext.SECURITY, ThreadContext.APPLICATION })
ManagedExecutor executor1;
... // in some other bean
@Inject
void setCompletableFuture(@NamedInstance("myExec") ManagedExecutor executor2) {
completableFuture = executor2.newIncompleteFuture();
}
... // in yet another bean
@Inject
@NamedInstance("myExec")
ManagedExecutor executor3;This specification defines a convention for defining properties in MicroProfile Config that override configuration attributes of ManagedExecutor and ThreadContext instances that are produced by the container. The convention for the property name is the fully qualified class name of the injection point, combined with the injection point field name or method name with parameter position, annotation name, and annotation attribute, with the / character as delimiter.
The following example shows one injection point for ManagedExecutor, which is the first parameter of the setCompletableFuture method, and another injection point for ThreadContext, which is the appContext field.
package org.eclipse.microprofile.example;
import org.eclipse.microprofile.concurrent.ManagedExecutor;
import org.eclipse.microprofile.concurrent.ThreadContext;
import org.eclipse.microprofile.concurrent.ThreadContextConfig;
import java.util.concurrent.CompletableFuture;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
@ApplicationScoped
public class ExampleBean {
CompletableFuture<Long> completableFuture;
@Inject
void setCompletableFuture(ManagedExecutor exec) {
completableFuture = exec.newIncompleteFuture();
}
@Inject @ThreadContextConfig(propagated = ThreadContext.APPLICATION,
cleared = ThreadContext.TRANSACTION,
unchanged = ThreadContext.ALL_REMAINING)
ThreadContext appContext;
}The following MicroProfile config properties could be used to override specific configuration attributes of these instances,
org.eclipse.microprofile.example.ExampleBean/setCompletableFuture/1/ManagedExecutorConfig/maxAsync=5
org.eclipse.microprofile.example.ExampleBean/setCompletableFuture/1/ManagedExecutorConfig/maxQueued=20
org.eclipse.microprofile.example.ExampleBean/appContext/ThreadContextConfig/cleared=Security,Transaction