Chapter 13. Persistence and Transactions

When you start a process, Red Hat JBoss BPM Suite creates a process instance, which represents the execution of the process in the specific context. For example, when you start a process that specifies how to process a sales order, Red Hat JBoss BPM Suite creates a process instance for each order. Process instances contain all the related information and minimal runtime state required to continue the execution at any time. However, it does not include process instance logs unless needed for execution of the process instance.

You can make the runtime state of an executing process persistent, for example, in a database. This allows you to restore the state of execution of all running processes in case of failure, or to temporarily remove running instances from memory and restore them later. Red Hat JBoss BPM Suite allows you to plug in different persistence strategies. Note that process instances are not persistent by default.

When you configure the Red Hat JBoss BPM Suite engine to use persistence, it automatically stores the runtime state in a database without further prompting. When you invoke the engine, it ensures that all changes are stored at the end of that invocation. If you encounter a failure and restore the engine from the database, do not manually resume the execution. Process instances automatically resume execution if they are triggered.

Inexperienced users should not directly access and modify database tables containing runtime persistence data. Changes in the runtime state of process instances which are not done by the engine may have unexpected results. If you require information about the current execution state of a process instance, use the history log.

Binary persistence, or marshaling, converts the state of the process instance into a binary dataset. Binary persistence is a mechanism used to store and retrieve information persistently. The same mechanism is also applied to the session state and work item states.

When you enable persistence of a process instance:

The session can also store other forms of state, such as the state of timer jobs, or data required for business rules evaluation. Session state is stored separately as a binary dataset along with the ID of the session and metadata. You can restore the session state by reloading a session with given ID. Use ksession.getId() to get the session ID.

Each entity of the data model is described below.

Figure 13.1. Data Model

The SessionInfo entity contains the state of the (knowledge) session in which the process instance is running.

The ProcessInstanceInfo entity contains the state of the process instance.

The EventTypes entity contains information about events that a process instance will undergo or has undergone.

The WorkItemInfo entity contains the state of a work item.

The CorrelationKeyInfo entity contains information about correlation keys assigned to the given process instance. This table is optional. Use it only when you require correlation capabilities.

The CorrelationPropertyInfo entity contains information about correlation properties for a correlation key assigned the process instance.

The ContextMappingInfo entity contains information about the contextual information mapped to a KieSession. This is an internal part of RuntimeManager and can be considered optional when RuntimeManager is not used.

During the process engine execution, the state of a process instance is stored in safe points. When you execute a process instance, the engine continues the execution until there are no more actions to be performed. That is, the process instance has been completed, aborted, or is in the wait state in all of its paths. At that point, the engine has reached the next safe state, and the state of the process instance (and all other process instances that it affected) is stored persistently.

The jbpm-audit module contains an event listener that stores process-related information in a database using Java Persistence API (JPA). The data model contains the following entities:

All audit data model entities contain following elements:

To log process history in a database, register a logger in your session:

EntityManagerFactory emf = ...;
StatefulKnowledgeSession ksession = ...;
AbstractAuditLogger auditLogger = AuditLoggerFactory.newJPAInstance(emf);
ksession.addProcessEventListener(auditLogger);

// Invoke methods on your session here.

Modify persistence.xml to specify a database. You need to include audit log classes as well (ProcessInstanceLog, NodeInstanceLog, and VariableInstanceLog). See the example:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>

  <persistence
    version="2.0"
    xsi:schemaLocation="
      http://java.sun.com/xml/ns/persistence
      http://java.sun.com/xml/ns/persistence/persistence_2_0.xsd
      http://java.sun.com/xml/ns/persistence/orm
      http://java.sun.com/xml/ns/persistence/orm_2_0.xsd"
    xmlns="http://java.sun.com/xml/ns/persistence"
    xmlns:orm="http://java.sun.com/xml/ns/persistence/orm"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

    <persistence-unit name="org.jbpm.persistence.jpa" transaction-type="JTA">
      <provider>org.hibernate.ejb.HibernatePersistence</provider>
      <jta-data-source>jdbc/jbpm-ds</jta-data-source>
      <mapping-file>META-INF/JBPMorm.xml</mapping-file>

      <class>org.drools.persistence.info.SessionInfo</class>
      <class>org.jbpm.persistence.processinstance.ProcessInstanceInfo</class>
      <class>org.drools.persistence.info.WorkItemInfo</class>
      <class>org.jbpm.persistence.correlation.CorrelationKeyInfo</class>
      <class>org.jbpm.persistence.correlation.CorrelationPropertyInfo</class>
      <class>org.jbpm.runtime.manager.impl.jpa.ContextMappingInfo</class>
      <class>org.jbpm.process.audit.ProcessInstanceLog</class>
      <class>org.jbpm.process.audit.NodeInstanceLog</class>
      <class>org.jbpm.process.audit.VariableInstanceLog</class>

      <properties>
        <property name="hibernate.dialect" value="org.hibernate.dialect.H2Dialect"/>
        <property name="hibernate.max_fetch_depth" value="3"/>
        <property name="hibernate.hbm2ddl.auto" value="update"/>
        <property name="hibernate.show_sql" value="true"/>
        <property name="hibernate.transaction.jta.platform"
                  value="org.hibernate.service.jta.platform.internal.BitronixJtaPlatform"/>
      </properties>
    </persistence-unit>
  </persistence>

Synchronous storing of history logs and runtime data in one database may be undesirable due to performance reasons. In that case, you can use JMS logger to send data into a JMS queue instead of directly storing it in a database. You can also configure it to be transactional in order to avoid issues with inconsistent data, for example when the process engine transaction is reversed.

Example configuration of JMS queue:

ConnectionFactory factory = ...;
Queue queue = ...;
StatefulKnowledgeSession ksession = ...;
Map<String, Object> jmsProps = new HashMap<String, Object>();

jmsProps.put("jbpm.audit.jms.transacted", true);
jmsProps.put("jbpm.audit.jms.connection.factory", factory);
jmsProps.put("jbpm.audit.jms.queue", queue);

AbstractAuditLogger auditLogger =
  AuditLoggerFactory.newInstance(Type.JMS, session, jmsProps);
ksession.addProcessEventListener(auditLogger);

// Invoke methods of your session here.

Process and task variables are stored as string (similar to variable.toString()) in audit tables by default. This is not always efficient, for example, when you need to query by the process or task instance variables:

public class Person implements Serializable {

  private static final long serialVersionUID = -5172443495317321032L;
  private String name;
  private int age;

  public Person(String name, int age) {
    this.name = name;
    this.age = age;
  }

  public String getName() {
    return name;
  }

  public void setName(String name) {
    this.name = name;
  }

  public int getAge() {
    return age;
  }

  public void setAge(int age) {
    this.age = age;
  }

  @Override
  public String toString() {
    return "Person [name=" + name + ", age=" + age + "]";
  }
}

In this example, when you want to query all the people with certain age, querying becomes inefficient.

Thus, variable audit is based on VariableIndexer, which extracts relevant parts of the variables that will be stored in audit log:

/**
* Variable indexer that allows to transform variable instance
* into other representation (usually String) to be able to use it for queries.
*
* @param <V> type of the object that will represent indexed variable
*/

public interface VariableIndexer<V> {

 /**
  * Tests if given variable shall be indexed by this indexer.
  *
  * NOTE: Only one indexer can be used for given variable.
  *
  * @param	variable  variable to be indexed
  * @return	true      if variable should be indexed with this indexer
  */

  boolean accept(Object variable);

 /**
  * Performs index/transform operation of the variable.
  * Result of this operation can be either single value
  * or list of values to support complex type separation.
  * For example, when variable is of type Person that has name,
  * address, and phone, indexer could build three entries
  * out of it to represent individual fields:
  *
  * person  = person.name
  * address = person.address.street
  * phone   = person.phone
  *
  * That will allow more advanced queries to be used to find
  * relevant entries.
  *
  * @param	name      name of the variable
  * @param	variable  actual variable value
  * @return
  */

  List<V> index(String name, Object variable);
}

The default indexer (that is indexer accepting toString()) produces a single audit entry for a single variable. However, you can create a custom indexer which indexes variables into separate audit entries:

public class PersonTaskVariablesIndexer implements TaskVariableIndexer {

  @Override
  public boolean accept(Object variable) {
    if (variable instanceof Person) {
      return true;
    }

    return false;
  }

  @Override
  public List<TaskVariable> index(String name, Object variable) {
    Person person = (Person) variable;
    List<TaskVariable> indexed = new ArrayList<TaskVariable>();

    TaskVariableImpl personNameVar = new TaskVariableImpl();
    personNameVar.setName("person.name");
    personNameVar.setValue(person.getName());

    indexed.add(personNameVar);

    TaskVariableImpl personAgeVar = new TaskVariableImpl();
    personAgeVar.setName("person.age");
    personAgeVar.setValue(person.getAge()+"");

    indexed.add(personAgeVar);

    return indexed;
  }
}

This allows you to search all the process instances or tasks that contain the person instance of age 34 by querying for:

You can build indexers for both process and task variables. They are supported by different interfaces because they produce different type of objects representing audit view of the variable. To create a custom indexer, follow these steps:

The ServiceLoader service registers indexers. When you start indexing, all the registered indexers are examined. If no applicable indexer is found, the default indexer (toString() based) is used.

To use persistence, add necessary dependencies to the classpath of your application. If you are using Red Hat JBoss Development Studio with Red Hat JBoss BPM Suite runtime default configuration, all necessary dependencies are already present for the default persistence configuration. Otherwise, ensure that the necessary JAR files are added to your Red Hat JBoss BPM Suite runtime directory.

Following is a list of dependencies for the default combination with Hibernate as the JPA persistence provider, an H2 in-memory database, and Bitronix for JTA-based transaction management. Dependencies needed for your project will vary depending on your solution configuration.

jbpm-persistence-jpa.jar file is necessary for saving the runtime state. Therefore, always make sure it is available in your project.

Use JPAKnowledgeService to create a knowledge session based on a knowledge base, a knowledge session configuration (if necessary), and the environment. Ensure that the environment contains a reference to your Entity Manager Factory. For example:

// Create the entity manager factory and register it in the environment:
EntityManagerFactory emf =
  Persistence.createEntityManagerFactory("org.jbpm.persistence.jpa");
Environment env = KnowledgeBaseFactory.newEnvironment();
env.set(EnvironmentName.ENTITY_MANAGER_FACTORY, emf);

// Create a new knowledge session that uses JPA to store the runtime state:
StatefulKnowledgeSession ksession =
  JPAKnowledgeService.newStatefulKnowledgeSession(kbase, null, env);
int sessionId = ksession.getId();

// Invoke methods on your session here:
ksession.startProcess("MyProcess");
ksession.dispose();

Additionally, you can use JPAKnowledgeService to recreate a session based on a specific session ID. For example:

// Recreate the session from database using the sessionId:

ksession = JPAKnowledgeService.loadStatefulKnowledgeSession(sessionId, kbase, null, env);

Note that only the minimal state that is required to continue execution of the process instance is saved. You cannot retrieve information related to already executed nodes if that information is no longer necessary. To search for history-related information, use the history log.

Add persistence.xml to META-INF to configure JPA. Following example uses Hibernate and H2 database:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<persistence
  version="2.0"
  xsi:schemaLocation="
    http://java.sun.com/xml/ns/persistence
    http://java.sun.com/xml/ns/persistence/persistence_2_0.xsd
    http://java.sun.com/xml/ns/persistence/orm
    http://java.sun.com/xml/ns/persistence/orm_2_0.xsd"
  xmlns="http://java.sun.com/xml/ns/persistence"
  xmlns:orm="http://java.sun.com/xml/ns/persistence/orm"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

  <persistence-unit name="org.jbpm.persistence.jpa" transaction-type="JTA">
    <provider>org.hibernate.ejb.HibernatePersistence</provider>
    <jta-data-source>jdbc/jbpm-ds</jta-data-source>
    <mapping-file>META-INF/JBPMorm.xml</mapping-file>

    <class>org.drools.persistence.info.SessionInfo</class>
    <class>org.jbpm.persistence.processinstance.ProcessInstanceInfo</class>
    <class>org.drools.persistence.info.WorkItemInfo</class>
    <class>org.jbpm.persistence.correlation.CorrelationKeyInfo</class>
    <class>org.jbpm.persistence.correlation.CorrelationPropertyInfo</class>
    <class>org.jbpm.runtime.manager.impl.jpa.ContextMappingInfo</class>

    <properties>
      <property name="hibernate.dialect" value="org.hibernate.dialect.H2Dialect"/>
      <property name="hibernate.max_fetch_depth" value="3"/>
      <property name="hibernate.hbm2ddl.auto" value="update"/>
      <property name="hibernate.show_sql" value="true"/>
      <property name="hibernate.transaction.jta.platform"
                value="org.hibernate.service.jta.platform.internal.BitronixJtaPlatform"/>
    </properties>
  </persistence-unit>
</persistence>

In this example, persistence.xml refers to a data source called jdbc/jbpm-ds. If you run your application in an application server, these containers typically allow you to use custom configure file for the data sources. See your application server documentation for further details.

Following example shows you how to set up a data source:

PoolingDataSource ds = new PoolingDataSource();

ds.setUniqueName("jdbc/jbpm-ds");
ds.setClassName("bitronix.tm.resource.jdbc.lrc.LrcXADataSource");
ds.setMaxPoolSize(3);
ds.setAllowLocalTransactions(true);
ds.getDriverProperties().put("user", "sa");
ds.getDriverProperties().put("password", "sasa");
ds.getDriverProperties().put("URL", "jdbc:h2:mem:jbpm-db");
ds.getDriverProperties().put("driverClassName", "org.h2.Driver");
ds.init();