Jakarta Concurrency

using CDI beans as tasks.

Specification: Jakarta Concurrency

Version: 3.1

Status: FINAL

Release: June 21, 2024
<resource-env-ref>
  <description>
    This executor is used for the application’s reporter task.
    This executor has the following requirements:
    Context Info: Local Namespace
  </description>
  <resource-env-ref-name>
    concurrent/LongRunningTasksExecutor
  </resource-env-ref-name>
  <resource-env-ref-type>
    jakarta.enterprise.concurrent.ManagedExecutorService
  </resource-env-ref-type>
</resource-env-ref>
public class ReporterTask implements Runnable {
  String reportName;

  public ReporterTask(String reportName) {
    this.reportName = reportName;
  }

public void run() {
  // Run the named report
  if("TransactionReport".equals(reportName)) {
    runTransactionReport();
  }
  else if("SummaryReport".equals(reportName)) {
    runSummaryReport();
  }
}

DataSource ds = …;

void runTransactionReport() {
  try (Connection con = ds.getConnection(); ...) {

    // Read/Write the data using our connection.
    ...
    // Commit.
    con.commit();
  }
}
public class AppServlet extends HTTPServlet implements Servlet {

  // Cache our executor instance
  @Resource(name=”concurrent/LongRunningTasksExecutor”)
  ManagedExecutorService mes;

  protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {

    // Get the name of the report to run from the input params...
    // Assemble the header for the response.
    // Create a task instance
    ReporterTask reporterTask = new ReporterTask(reportName);

    // Submit the task to the ManagedExecutorService
    Future reportFuture = mes.submit(reporterTask);

    // Cache the future somewhere (like the client's session)
    // The client can then poll the servlet to determine
    // the status of the report.
    ...

    // Tell the user that the report has been submitted.
    ...
  }
}
<resource-env-ref>
  <description>
    This executor is used for the application’s builder tasks.
    This executor has the following requirements:
    Context Info: Local Namespace, Security
  </description>
  <resource-env-ref-name>
    concurrent/BuilderExecutor
  </resource-env-ref-name>
  <resource-env-ref-type>
    jakarta.enterprise.concurrent.ManagedExecutorService
  </resource-env-ref-type>
</resource-env-ref>
public class AccountTask implements Callable<AccountInfo>, ManagedTask
{
  // The ID of the request to report on demand.
  String reqID;

  String accountID;
  Map<String, String> execProps;

  public AccountTask(String reqID, String accountID) {
    this.reqID=reqID;
    this.accountID=accountID;
    execProps = new HashMap<>();
    execProps.put(ManagedTask.IDENTITY_NAME, getIdentityName());
  }

  public AccountInfo call() {

    // Retrieve account info for the account from some persistent store
    AccountInfo info = ...;
    return info;
  }

  public String getIdentityName() {
    return "AccountTask: ReqID=" + reqID + ", Acct=" + accountID;
  }

  public Map<String, String> getExecutionProperties() {
    return execProps;
  }

  public ManagedTaskListener getManagedTaskListener() {
    return null;
  }
}

public class InsuranceTask implements Callable<InsuranceInfo>, ManagedTask {

  // The ID of the request to report on demand.
  String reqID

  String accountID;
  Map<String, String> execProps;

  public InsuranceTask (String reqID, String accountID) {
    this.reqID=reqID;
    this.accountID=accountID;
    execProps = new HashMap<>();

    execProps.put(ManagedTask.IDENTITY_NAME, getIdentityName());
  }

  public InsuranceInfo call() {
    // Retrieve the insurance info for the account from some persistent store
    InsuranceInfo info = ...;
    return info;
  }

  public String getIdentityName() {
    return "InsuranceTask: ReqID=" + reqID + ", Acct=" + accountID;
  }

  public Map<String, String> getExecutionProperties()
    return execProps;
  }

  public ManagedTaskListener getManagedTaskListener() {
    return null;
  }
}
public class AppServlet extends HttpServlet implements Servlet {

  // Retrieve our executor instance.
  @Resource(name=”concurrent/BuilderExecutor”)
  ManagedExecutorService mes;

  protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
    // Get our arguments from the request (accountNumber and
    // requestID, in this case.
    // Assemble the header for the response.
    // Create and submit the task instances

    Future<AccountInfo> acctFuture = mes.submit(new AccountTask(reqID, accountID));

    Future<InsuranceInfo> insFuture = mes.submit (new InsuranceTask(reqID, accountID));

    // Wait for the results.
    AccountInfo accountInfo = acctFuture.get();
    InsuranceInfo insInfo = insFuture.get();

    // Process the results
  }
}
public class TranTask implements Runnable {

  UserTransaction ut = …;

  public void run() {

    // Start a transaction
    ut.begin();

    // Invoke an Jakarta Enterprise Bean
    ...

    //
    Update a database using an XA capable JDBC DataSource
    ...

    // Commit the transaction
    ut.commit();
  }
}
<resource-env-ref>
  <description>
    This executor is used for the application’s logger task.
    This executor has the following requirements:
    Context Info: Local Namespace
  </description>
  <resource-env-ref-name>
    concurrent/ScheduledLoggerExecutor
  </resource-env-ref-name>
  <resource-env-ref-type>
    jakarta.enterprise.concurrent.ManagedScheduledExecutorService
  </resource-env-ref-type>
</resource-env-ref>
public class LoggerTimer implements Runnable {
  DataSource ds = ...;

  public void run() {
    logEvents(getData(), ds);
  }

  void logEvents(Collection data, DataSource ds) {

    // Iterate through the data and log each row.
    for (...) {
      try (Connection con = ds.getConnection(); ...) {

        // Write the data using our connection.
        ...

        // Commit.
        con.commit();
      }
    }
  }
}
public class CtxListener implements ServletContextListener {

  Future loggerHandle = null;

  @Resource(name=”concurrent/ScheduledLoggerExecutor”)
  ManagedScheduledExecutorService mes;

  public void contextInitialized(ServletContextEvent scEvent) {
    LoggerTimer logger = new LoggerTimer();
    loggerHandle = mes.scheduleAtFixedRate(logger, 5, TimeUnit.SECONDS);
  }

  public void contextDestroyed(ServletContextEvent scEvent) {

    // Cancel and interrupt our logger task
    if(loggerHandle!=null) {
      loggerHandle.cancel(true);
    }
  }
}
@Singleton
public class ExecutorAccessor {

  private ExecutorService threadPoolExecutor = null;

  @Resource(name="concurrent/ThreadFactory")
  ManagedThreadFactory threadFactory;

  @PostConstruct
  public void postConstruct() {
    threadPoolExecutor = new ThreadPoolExecutor( 5, 10, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), threadFactory);
  }

  public ExecutorService getThreadPool() {
    return threadPoolExecutor;
  }
}
<resource-env-ref>
  <description>
    This ThreadFactory is used for the singleton ThreadPoolExecutor.
    Priority: Normal
    Context Info: NA
  </description>

  <resource-env-ref-name>
    concurrent/ThreadFactory
  </resource-env-ref-name>

  <resource-env-ref-type>
    jakarta.enterprise.concurrent.ManagedThreadFactory
  </resource-env-ref-type>
</resource-env-ref>

<resource-env-ref>
  <description>
    This ContextService is used in conjunction with the custom
    ThreadPoolExecutor that the credit report component is using.
    This ContextService has the following requirements:
    Context Info: Local namespace, security
  </description>

  <resource-env-ref-name>
    concurrent/AllContexts
  </resource-env-ref-name>

  <resource-env-ref-type>
    jakarta.enterprise.concurrent.ContextService
  </resource-env-ref-type>
</resource-env-ref>
public class CreditScoreTask implements Callable<Long> {
  private long taxID;
  private int agency;

  public CreditScoreTask(long taxID, int agency) {
    this.taxID = taxID;
    this.agency = agency;
  }

  public Long call() {
    // Log the request in a database using the identity of the user.
    // Use the local namespace to locate the datasource
    Subject currentSubject = Subject.getSubject(AccessController.getContext());
    logCreditAccess(currentSubject, taxID, agency);

    // Use Web Services to retrieve the credit score from the
    // specified agency.
    return getCreditScore(taxID, agency);
  }

  ...
}
class LoanCheckerBean {
  @Resource(name="concurrent/AllContexts")
  ContextService ctxSvc;

  @EJB private ExecutorAccessor executorAccessor;

  public long[] getCreditScores(long taxID) {
    // Retrieve our singleton threadpool, but wrap it in
    // a ExecutorCompletionService
    ExecutorCompletionService<Long> threadPool = new ExecutorCompletionService<Long>(executorAccessor.getThreadPool());

    // Use this thread to retrieve one credit score, and
    // use two other threads to process the other two scores.
    // Since we are using a custom executor and
    // because our tasks depend upon the context in which this
    // method is running, we use a contextual task.
    CreditScoreTask agency1 = new CreditScoreTask(taxID, 1);

    Callable<Long> agency2 = ctxSvc.createContextualProxy( new CreditScoreTask(taxID, 2), Callable.class));

    Callable<Long> agency3 = ctxSvc.createContextualProxy ( new CreditScoreTask(taxID, 3), Callable.class));

    threadPool.submit(agency2);
    threadPool.submit(agency3);

    long[] scores = {0,0,0};
    try {
      // Retrieve one credit score on this thread.
      scores[0] = agency1.call();

      // Retrieve the other two credit scores
      scores[1] = threadPool.take().get();
      scores[2] = threadPool.take().get();
    } catch (InterruptedException e) {
      // The app may be shutting down.
    } catch (ExecutionException e) {
      // There was an error retrieving one of the asynch scores.
    }
    return scores;
  }
}
<resource-env-ref>
  <description>
    This ManagedThreadFactory is used to create a thread for for the
    application’s logger task.
    This ManagedThreadFactory has the following requirements:
    Context Info: Local Namespace
  </description>

  <resource-env-ref-name>
    concurrent/LoggerThreadFactory
  </resource-env-ref-name>

  <resource-env-ref-type>
    jakarta.enterprise.concurrent.ManagedThreadFactory
  </resource-env-ref-type>
</resource-env-ref>
public class LoggerTask implements Runnable {

  DataSource ds = ...;

  public void run() {
    // Wait for data and log it.
    while (!Thread.interrupted()) {
      logEvents(getData(), ds);
    }
  }

  void logEvents(Collection data, DataSource ds) {
    // Iterate through the data and log each row.
    for (...) {
      try (Connection con = ds.getConnection();... {

        // Write the data using our connection.
        ...

        // Commit.
        con.commit();
      }
    }
  }
}
public class CtxListener implments ServletContextListener {

  Thread loggerThread = null;

  @Resource(name=”concurrent/LoggerThreadFactory”)
  ManagedThreadFactory threadFactory;

  public void contextInitialized(ServletContextEvent scEvent) {
    LoggerTask logger = new LoggerTask();
    Thread loggerThread = threadFactory.newThread(logger);
    loggerThread.start();
  }

  public void contextDestroyed(ServletContextEvent scEvent) {
    // Interrupt our logger task since it is no longer available.
    // Note: The server will do this for us as well.
    if (loggerThread!=null) {
      loggerThread.interrupt();
    }
  }
}
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

import jakarta.inject.Qualifier;

@Qualifier
@Retention(RetentionPolicy.RUNTIME)
@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.TYPE })
public @interface MaxAsync5 {
}
@ManagedExecutorDefinition(
        name = "java:comp/concurrent/MyExecutor",
        qualifiers = Max5Async.class,
        maxAsync = 5)
public class MyServlet extends HttpServlet {
   @Inject
   @Max5Async
   ManagedExecutorService executor;

   ...
}
package example.jakarta.concurrency.priority;

import jakarta.enterprise.concurrent.spi.ThreadContextProvider;
import jakarta.enterprise.concurrent.spi.ThreadContextSnapshot;
import java.util.Map;

public class ThreadPriorityContextProvider implements ThreadContextProvider {
   public String getThreadContextType() {
      return "ThreadPriority";
   }

   public ThreadContextSnapshot currentContext(Map<String, String> props) {
      return new ThreadPrioritySnapshot(Thread.currentThread().getPriority());
   }

   public ThreadContextSnapshot clearedContext(Map<String, String> props) {
      return new ThreadPrioritySnapshot(Thread.NORM_PRIORITY);
   }
}
package example.jakarta.concurrency.priority;

import jakarta.enterprise.concurrent.spi.ThreadContextRestorer;
import jakarta.enterprise.concurrent.spi.ThreadContextSnapshot;
import java.util.concurrent.atomic.AtomicBoolean;

public class ThreadPrioritySnapshot implements ThreadContextSnapshot {
   final int priority;

   ThreadPrioritySnapshot(int priority) {
      this.priority = priority;
   }

   public ThreadContextRestorer begin() {
      Thread thread = Thread.currentThread();
      int priorityToRestore = thread.getPriority();
      AtomicBoolean restored = new AtomicBoolean();

      ThreadContextRestorer contextRestorer = () -> {
         if (restored.compareAndSet(false, true))
            thread.setPriority(priorityToRestore);
         else
            throw new IllegalStateException();
      };

      thread.setPriority(priority);

      return contextRestorer;
   }
}
META-INF/services/jakarta.enterprise.concurrent.spi.ThreadContextProvider
example.jakarta.concurrency.priority.ThreadPriorityContextProvider
import jakarta.annotation.Resource;
import jakarta.enterprise.concurrent.ContextServiceDefinition;
import jakarta.enterprise.concurrent.ManagedExecutorDefinition;
import jakarta.enterprise.concurrent.ManagedExecutorService;
import jakarta.servlet.ServletException;
import jakarta.servlet.http.HttpServlet;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import java.io.IOException;

@ContextServiceDefinition(
        name = "java:module/concurrent/PriorityContext",
        propagated = "ThreadPriority")
@ManagedExecutorDefinition(
        name = "java:module/concurrent/PriorityExec",
        context = "java:module/concurrent/PriorityContext")
public class MyServlet extends HttpServlet {
   @Resource(lookup = "java:module/concurrent/PriorityExec")
   ManagedExecutorService executor;

   protected void doGet(HttpServletRequest req, HttpServletResponse res)
      throws ServletException, IOException {

      Thread.currentThread().setPriority(3);

      executor.runAsync(() -> {
         System.out.println("Running with priority of " +
            Thread.currentThread().getPriority());
      });
   }
}
public class ProductRecommendations {
   @Asynchronous(executor = "java:module/env/concurrent/myExecutorRef")
   public CompletableFuture<Set<Item>> findSimilar(Cart cart, History h) {
      Set<Item> combined = new LinkedHashSet<Item>();
      for (Item item : cart.items())
         combined.addAll(item.similar());
      for (Item item : h.recentlyViewed(3))
         combined.addAll(item.similar());
      combined.removeAll(cart.items());

      try (Connection con = ((DataSource) InitialContext.doLookup(
                            "java:comp/env/jdbc/ds1")).getConnection()) {
         PreparedStatement stmt = con.prepareStatement(CHECK_AVAILABILITY);
         for (Item item : combined) {
            ... Remove if the similar item is unavailable
         }
      } catch (NamingException | SQLException x) {
         throw new CompletionException(x);
      }
      return Asynchronous.Result.complete(combined);
   }
}
public class CheckoutServlet extends HttpServlet {
   @Inject
   ProductRecommendations recommendations;

   @Resource(name=”java:comp/env/jdbc/ds1”, lookup="jdbc/ds1")
   DataSource ds;

   public void doGet(HttpServletRequest req, HttpServletResponse resp)
      throws ServletException {
      ...
      recommendations.findSimilar(cust.getCart(), cust.getHistory())
                     .thenAccept(recommended -> {
                        ... Update page with recommendations
                     });
      ...
   }
}

Jakarta Concurrency

Eclipse Foundation Specification License - v1.1

Disclaimers

Jakarta Concurrency Specification, Version 3.1

1. Introduction

1.1. Overview

1.2. Goals of this specification

1.3. Other Java Platform and Jakarta Specifications

1.4. Concurrency Utilities for Java EE Expert Group at the JCP

1.5. Document Conventions

2. Overview

2.1. Container-Managed vs. Unmanaged Threads

2.2. Application Integrity

2.3. Container Thread Context

2.3.1. Contextual Invocation Points

2.3.1.1. Flow Contextual Invocation Points

2.3.1.2. Optional Contextual Invocation Points

2.3.2. Contextual Objects and Tasks

2.3.2.1. Tasks and Jakarta Contexts and Dependency Injection (CDI)

2.4. Usage with Jakarta Connectors

2.5. Security

3. Managed Objects

3.1. ManagedExecutorService

3.1.1. Application Component Provider’s Responsibilities

3.1.1.1. Usage Example

Reporter Task

Resource Environment Reference - Reporter Task

Task Definition – Reporter Task

Task Submission – Reporter Task

Builder Tasks

Resource Environment Reference – Builder Tasks

Task Definition – Builder Tasks

Task Invocation – Builder Tasks

3.1.2. Application Assembler’s Responsibilities

3.1.3. Deployer’s Responsibilities

3.1.4. Jakarta EE Product Provider’s Responsibilities

3.1.4.1. ManagedExecutorService Configuration Attributes

3.1.4.2. Configuration Examples

Typical Thread Pool

Thread Pool for Long-Running Tasks

OLTP Thread Pool

3.1.4.3. Default ManagedExecutorService

3.1.5. System Administrator’s Responsibilities

3.1.6. Lifecycle

3.1.6.1. Jakarta EE Product Provider Requirements

3.1.7. Quality of Service

3.1.8. Transaction Management

3.1.8.1. Jakarta EE Product Provider Requirements

3.1.8.2. Application Component Provider’s Requirements

UserTransaction Usage Example

3.2. ManagedScheduledExecutorService

3.2.1. Application Component Provider’s Responsibilities

3.2.1.1. Usage Example

Logger Timer Task

Resource Environment Reference

Task Definition

Task Submission

3.2.2. Application Assembler’s Responsibilities

3.2.3. Deployer’s Responsibilities

3.2.4. Jakarta EE Product Provider’s Responsibilities

3.2.4.1. ManagedScheduledExecutorService Configuration Attributes

3.2.4.2. Configuration Examples

Typical Timer

3.2.4.3. Default ManagedScheduledExecutorService

3.2.5. System Administrator’s Responsibilities

3.2.6. Lifecycle

3.2.6.1. Jakarta EE Product Provider Requirements

3.2.7. Quality of Service

3.2.8. Transaction Management

3.2.8.1. Jakarta EE Product Provider Requirements

3.2.8.2. Application Component Provider’s Requirements

3.3. ContextService

3.3.1. Application Component Provider’s Responsibilities

3.3.1.1. Usage Example

Custom ExecutorService

ExecutorService Singleton

CreditReport Task

Resource Environment References

Task Definition

Task Invocation