Working with Events

<bpmn2:itemDefinition id="CorrelationPropertyId_Type" structureRef="String" />
<bpmn2:correlationProperty id="CorrelationPropertyId" name="Correlation Property 1" type="CorrelationPropertyId_Type">
  <bpmn2:correlationPropertyRetrievalExpression
    id="_01E88DC7-3BDD-48E3-BCF9-39AD0772B2AB"
    messageRef="_GlqmEIpdED2dBNv7eRNehQ"
  >
    <bpmn2:messagePath
      id="_F1B5A911-766C-4E07-932E-A7C9BC825274"
      evaluatesToTypeRef="_6404189C-98F1-471A-899F-4A66C9E9BB0E"
      language="java"
    >id</bpmn2:messagePath>
  </bpmn2:correlationPropertyRetrievalExpression>
</bpmn2:correlationProperty>

<bpmn2:correlationKey id="CorrelationKeyId" name="Correlation Key Name">
  <bpmn2:correlationPropertyRef>CorrelationPropertyId</bpmn2:correlationPropertyRef>
</bpmn2:correlationKey>

# Kafka broker URL
kafka.bootstrap.servers=localhost:9092

# Incoming channel (for Start/Catch/Boundary events)
mp.messaging.incoming.<channel-name>.connector=smallrye-kafka
mp.messaging.incoming.<channel-name>.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.<channel-name>.auto.offset.reset=earliest

# Outgoing channel (for Throw/End events)
mp.messaging.outgoing.<channel-name>.connector=smallrye-kafka
mp.messaging.outgoing.<channel-name>.value.serializer=org.apache.kafka.common.serialization.StringSerializer

package org.kie.api.event.process;

import java.util.EventListener;

/**
 * A listener for events related to Process Instance execution.
 */
public interface ProcessEventListener extends EventListener {

    /**
     * This listener method is invoked immediately before a Process Instance is started.
     * @param event
     */
    void beforeProcessStarted(ProcessStartedEvent event);

    /**
     * This listener method is invoked immediately after a Process Instance has been started.
     * @param event
     */
    void afterProcessStarted(ProcessStartedEvent event);

    /**
     * This listener method is invoked immediately before a Process Instance is completed (or aborted).
     * @param event
     */
    void beforeProcessCompleted(ProcessCompletedEvent event);

    /**
     * This listener method is invoked immediately after a Process Instance has been completed (or aborted).
     * @param event
     */
    void afterProcessCompleted(ProcessCompletedEvent event);

    /**
     * This listener method is invoked immediately before a node in a Process Instance is triggered
     * (which is when the node is entered, for example when an incoming connection triggers it).
     * @param event
     */
    void beforeNodeTriggered(ProcessNodeTriggeredEvent event);

    /**
     * This listener method is invoked immediately after a node in a Process Instance has been triggered
     * (which is when the node was entered, for example when an incoming connection triggered it).
     * @param event
     */
    void afterNodeTriggered(ProcessNodeTriggeredEvent event);

    /**
     * This listener method is invoked immediately before a node in a Process Instance is left
     * (which is when the node is completed, for example when it has performed the task it was
     * designed for).
     * @param event
     */
    void beforeNodeLeft(ProcessNodeLeftEvent event);

    /**
     * This listener method is invoked immediately after a node in a Process Instance has been left
     * (which is when the node was completed, for example when it performed the task it was
     * designed for).
     * @param event
     */
    void afterNodeLeft(ProcessNodeLeftEvent event);

    /**
     * This listener method is invoked immediately before the value of a process variable is changed.
     * @param event
     */
    void beforeVariableChanged(ProcessVariableChangedEvent event);

    /**
     * This listener method is invoked immediately after the value of a process variable has been changed.
     * @param event
     */
    void afterVariableChanged(ProcessVariableChangedEvent event);

    /**
     * This listener method is invoked immediately before the SLA of a Process/node Instance is violated.
     * @param event
     */
    default void beforeSLAViolated(SLAViolatedEvent event) {}

    /**
     * This listener method is invoked immediately after the SLA of a Process/node Instance has been violated.
     * @param event
     */
    default void afterSLAViolated(SLAViolatedEvent event) {}

    /**
     * This listener method is invoked when a signal is sent
     * @param event
     */
    default void onSignal(SignalEvent event) {}

    /**
     * This listener is triggered when a migration occurs
     * @param event
     */
    default void onMigration(ProcessMigrationEvent event) { }

    /**
     * This listener method is invoked when a message is sent
     * @param event
     */
    default void onMessage(MessageEvent event) {}

    /**
     * This listener method is invoked when an error is captured
     * @param event
     */
    default void onError (ErrorEvent event) {}

    /**
     * This listener method is invoked immediately after a Process Instance is modified.
     * @param event
     */
    default void onProcessStateChanged(ProcessStateChangeEvent event) {}

    /**
     * This listener method is invoked immediately after a node in a Process Instance is modified.
     * @param event
     */
    default void onNodeStateChanged(ProcessNodeStateChangeEvent event) {}
}

public interface ProcessEvent extends KieRuntimeEvent {
    ProcessInstance getProcessInstance();       // The process instance this event relates to
    Date getEventDate();                        // When the event was created
    default String getEventIdentity() {         // Identity that performed the event
        return null;
    }
}

WorkflowProcessInstance processInstance = event.getNodeInstance().getProcessInstance();
NodeType nodeType = event.getNodeInstance().getNode().getNodeType();

@ApplicationScoped
public class MyProcessEventListener implements ProcessEventListener {

    @Override
    public void afterVariableChanged(ProcessVariableChangedEvent event) {
        String variableId = event.getVariableId();
        Object oldValue = event.getOldValue();
        Object newValue = event.getNewValue();

        System.out.println("Variable '" + variableId + "' changed from "
            + oldValue + " to " + newValue);
    }

    // Implement other methods as needed or extend DefaultProcessEventListener
}

package org.kie.kogito.usertask;

import org.kie.kogito.usertask.events.UserTaskAssignmentEvent;
import org.kie.kogito.usertask.events.UserTaskAttachmentEvent;
import org.kie.kogito.usertask.events.UserTaskCommentEvent;
import org.kie.kogito.usertask.events.UserTaskDeadlineEvent;
import org.kie.kogito.usertask.events.UserTaskStateEvent;
import org.kie.kogito.usertask.events.UserTaskVariableEvent;

public interface UserTaskEventListener {

    /**
     * This listener method is invoked when a user task deadline event occurs.
     * @param event
     */
    default void onUserTaskDeadline(UserTaskDeadlineEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when a user task state changes.
     * The event contains both the old status and new status.
     * @param event
     */
    default void onUserTaskState(UserTaskStateEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when a user task assignment changes.
     * The event contains both the old users and new users.
     * @param event
     */
    default void onUserTaskAssignment(UserTaskAssignmentEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when a user task input variable changes.
     * The event contains both the old value and new value.
     * @param event
     */
    default void onUserTaskInputVariable(UserTaskVariableEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when a user task output variable changes.
     * The event contains both the old value and new value.
     * @param event
     */
    default void onUserTaskOutputVariable(UserTaskVariableEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when an attachment is added to a user task.
     * @param event
     */
    default void onUserTaskAttachmentAdded(UserTaskAttachmentEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when an attachment is deleted from a user task.
     * @param event
     */
    default void onUserTaskAttachmentDeleted(UserTaskAttachmentEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when an attachment is modified in a user task.
     * @param event
     */
    default void onUserTaskAttachmentChange(UserTaskAttachmentEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when a comment is modified in a user task.
     * @param event
     */
    default void onUserTaskCommentChange(UserTaskCommentEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when a comment is added to a user task.
     * @param event
     */
    default void onUserTaskCommentAdded(UserTaskCommentEvent event) {
        // nothing
    }

    /**
     * This listener method is invoked when a comment is deleted from a user task.
     * @param event
     */
    default void onUserTaskCommentDeleted(UserTaskCommentEvent event) {
        // nothing
    }
}

public interface UserTaskStateEvent extends UserTaskEvent {
    UserTaskState getNewStatus();  // Current state
    UserTaskState getOldStatus();  // Previous state
}

public interface UserTaskAssignmentEvent extends UserTaskEvent {
    String getAssignmentType();
    String[] getNewUsersId();  // New assigned users
    String[] getOldUsersId();  // Previously assigned users
}

public interface UserTaskVariableEvent extends UserTaskEvent {
    enum VariableEventType {
        INPUT,
        OUTPUT
    }

    String getVariableName();
    Object getOldValue();  // Previous value
    Object getNewValue();  // Current value
    VariableEventType getVariableType();
}

public interface UserTaskEvent {
    UserTask getUserTask();
    UserTaskInstance getUserTaskInstance();
    Date getEventDate();
    String getEventUser();
}

@ApplicationScoped
public class MyUserTaskEventListener implements UserTaskEventListener {

    @Override
    public void onUserTaskState(UserTaskStateEvent event) {
        String taskName = event.getUserTask().getTaskName();
        UserTaskState oldStatus = event.getOldStatus();
        UserTaskState newStatus = event.getNewStatus();

        System.out.println("Task '" + taskName + "' changed from "
            + (oldStatus != null ? oldStatus.getName() : "null")
            + " to " + newStatus.getName());
    }

    @Override
    public void onUserTaskInputVariable(UserTaskVariableEvent event) {
        String variableName = event.getVariableName();
        Object oldValue = event.getOldValue();
        Object newValue = event.getNewValue();

        System.out.println("Task input variable '" + variableName + "' changed from "
            + oldValue + " to " + newValue);
    }

    @Override
    public void onUserTaskAssignment(UserTaskAssignmentEvent event) {
        String[] oldUsers = event.getOldUsersId();
        String[] newUsers = event.getNewUsersId();
        String taskName = event.getUserTask().getTaskName();

        System.out.println("Task '" + taskName + "' assignment changed from "
            + (oldUsers != null ? String.join(",", oldUsers) : "null") + " to "
            + (newUsers != null ? String.join(",", newUsers) : "null"));
    }
}

package org.acme;

import org.kie.api.event.process.ProcessCompletedEvent;
import org.kie.api.event.process.ProcessNodeLeftEvent;
import org.kie.api.event.process.ProcessNodeTriggeredEvent;
import org.kie.api.event.process.ProcessStartedEvent;
import org.kie.api.event.process.ProcessVariableChangedEvent;
import org.kie.kogito.internal.process.event.DefaultKogitoProcessEventListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import jakarta.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class MyProcessEventListener extends DefaultKogitoProcessEventListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyProcessEventListener.class);

    @Override
    public void beforeProcessStarted(ProcessStartedEvent event) {
        LOGGER.info("beforeProcessStarted: processId={}",
                event.getProcessInstance().getProcessId());
    }

    @Override
    public void afterProcessStarted(ProcessStartedEvent event) {
        LOGGER.info("afterProcessStarted: processId={}, state={}",
                event.getProcessInstance().getProcessId(),
                event.getProcessInstance().getState());
    }

    @Override
    public void beforeProcessCompleted(ProcessCompletedEvent event) {
        LOGGER.info("beforeProcessCompleted: processId={}",
                event.getProcessInstance().getProcessId());
    }

    @Override
    public void afterProcessCompleted(ProcessCompletedEvent event) {
        LOGGER.info("afterProcessCompleted: processId={}, state={}",
                event.getProcessInstance().getProcessId(),
                event.getProcessInstance().getState());
    }

    @Override
    public void beforeNodeTriggered(ProcessNodeTriggeredEvent event) {
        LOGGER.info("beforeNodeTriggered: nodeName={}",
                event.getNodeInstance().getNodeName());
    }

    @Override
    public void afterNodeTriggered(ProcessNodeTriggeredEvent event) {
        LOGGER.info("afterNodeTriggered: nodeName={}",
                event.getNodeInstance().getNodeName());
    }

    @Override
    public void beforeNodeLeft(ProcessNodeLeftEvent event) {
        LOGGER.info("beforeNodeLeft: nodeName={}",
                event.getNodeInstance().getNodeName());
    }

    @Override
    public void afterNodeLeft(ProcessNodeLeftEvent event) {
        LOGGER.info("afterNodeLeft: nodeName={}",
                event.getNodeInstance().getNodeName());
    }

    @Override
    public void beforeVariableChanged(ProcessVariableChangedEvent event) {
        LOGGER.info("beforeVariableChanged: variableId={}, OLD_VALUE={}, NEW_VALUE={}",
                event.getVariableId(),
                event.getOldValue(),
                event.getNewValue());
    }

    @Override
    public void afterVariableChanged(ProcessVariableChangedEvent event) {
        LOGGER.info("afterVariableChanged: variableId={}, OLD_VALUE={}, NEW_VALUE={}",
                event.getVariableId(),
                event.getOldValue(),
                event.getNewValue());
    }
}

package org.acme;

import org.kie.kogito.usertask.UserTaskEventListener;
import org.kie.kogito.usertask.events.UserTaskAssignmentEvent;
import org.kie.kogito.usertask.events.UserTaskStateEvent;
import org.kie.kogito.usertask.events.UserTaskVariableEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import jakarta.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class MyUserTaskEventListener implements UserTaskEventListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyUserTaskEventListener.class);

    @Override
    public void onUserTaskState(UserTaskStateEvent event) {
        LOGGER.info("onUserTaskState: taskName={}, OLD_STATUS={}, NEW_STATUS={}",
                event.getUserTaskInstance().getTaskName(),
                event.getOldStatus(),
                event.getNewStatus());
    }

    @Override
    public void onUserTaskAssignment(UserTaskAssignmentEvent event) {
        LOGGER.info("onUserTaskAssignment: taskName={}, OLD_USERS={}, NEW_USERS={}",
                event.getUserTaskInstance().getTaskName(),
                event.getOldUsersId() != null ? String.join(",", event.getOldUsersId()) : "null",
                event.getNewUsersId() != null ? String.join(",", event.getNewUsersId()) : "null");
    }

    @Override
    public void onUserTaskInputVariable(UserTaskVariableEvent event) {
        LOGGER.info("onUserTaskInputVariable: variableName={}, OLD_VALUE={}, NEW_VALUE={}",
                event.getVariableName(),
                event.getOldValue(),
                event.getNewValue());
    }

    @Override
    public void onUserTaskOutputVariable(UserTaskVariableEvent event) {
        LOGGER.info("onUserTaskOutputVariable: variableName={}, OLD_VALUE={}, NEW_VALUE={}",
                event.getVariableName(),
                event.getOldValue(),
                event.getNewValue());
    }
}

@ApplicationScoped
public class ExternalSystemUserTaskListener implements UserTaskEventListener {

    @Override
    public void onUserTaskState(UserTaskStateEvent event) {
        UserTaskInstance task = event.getUserTaskInstance();

        // All data available in one object:
        task.getId();                // Task identifier
        task.getTaskName();          // Task name
        task.getStatus().getName();  // Current status
        task.getActualOwner();       // Assigned user
        task.getPotentialUsers();    // Potential owners (users)
        task.getPotentialGroups();   // Potential owners (groups)
        task.getExcludedUsers();     // Excluded users
        task.getAdminUsers();        // Business administrators (users)
        task.getAdminGroups();       // Business administrators (groups)
        task.getInputs();            // Input variables (Map)
        task.getOutputs();           // Output variables (Map)
        task.getAttachments();       // Task attachments
        task.getComments();          // Task comments
        task.getMetadata();          // Task metadata
        task.getSlaDueDate();        // SLA due date

        // Process context
        ProcessInfo info = task.getProcessInfo();
        info.getProcessInstanceId();
        info.getProcessId();
        info.getProcessVersion();

        // State transition info from event
        event.getOldStatus();        // Previous state
        event.getNewStatus();        // Current state
        event.getEventDate();        // When it happened
        event.getEventUser();        // Who triggered it
    }
}

@ApplicationScoped
public class JsonExportUserTaskListener implements UserTaskEventListener {

    @Inject
    ObjectMapper mapper;

    @Override
    public void onUserTaskState(UserTaskStateEvent event) {
        try {
            UserTaskInstance task = event.getUserTaskInstance();
            String json = mapper.writeValueAsString(task);
            LOGGER.info("Task JSON: {}", json);
            // Send to Elasticsearch, Kafka, etc.
        } catch (JsonProcessingException e) {
            LOGGER.error("Error serializing task", e);
        }
    }
}

{
  {
  "id": "71a23b77-98ee-4a57-ad6f-f532b39630c7",
  "userTaskId": "_B8C4F63C-81AD-4291-9C1B-84967277EEF6",
  "processInfo": {
    "processInstanceId": "18fc537b-567a-4904-9a26-1094f4e29822",
    "processId": "hiring",
    "processVersion": "1.0"
  },
  "status": { "terminate": null, "name": "Ready" },
  "actualOwner": "jdoe",
  "taskName": "HRInterview",
  "potentialUsers": ["jdoe"],
  "potentialGroups": ["hr-manager"],
  "adminUsers": [],
  "adminGroups": [],
  "excludedUsers": [],
  "inputs": {
    "offer": { "category": "Senior Software Engineer", "salary": 40450 },
    "candidate": { "name": "Jon", "lastName": "Snow", "email": "jon@snow.org", "experience": 5, "skills": ["Java", "Kogito", "Fencing"] },
    "approve": false
  },
  "outputs": {},
  "attachments": [],
  "comments": [],
  "metadata": { "ProcessType": "BPMN", "Lifecycle": "kogito", "Skippable": "false" }
}

@Override
public void onUserTaskState(UserTaskStateEvent event) {
    try {
        // External service call
        externalService.notify(event.getUserTaskInstance());
    } catch (Exception e) {
        LOGGER.error("Error in listener: {}", e.getMessage(), e);
        // Don't rethrow — allow the transaction to complete
    }
}

@Override
public void afterVariableChanged(ProcessVariableChangedEvent event) {
    try {
        auditService.recordChange(event);
    } catch (Exception e) {
        LOGGER.error("Error in listener: {}", e.getMessage(), e);
    }
}