Kafka Consumers

Availability of functionality depends on the license and may not be available in all deployments.

Purpose

Processes in Eximee may consist of steps that require asynchronous communication with external systems. For example: we send a request to verify documents, but this is time-consuming, so we will have to wait for a response. This task can be performed asynchronously in an external system, and it can notify the Eximee process via Kafka. Kafka consumers are used precisely to handle such responses. We use them to receive the message, pre-process the data contained in it, and notify Eximee BPMS to continue the process execution.

Mapping a Kafka topic to a Kafka Consumer

Every Kafka consumer must be linked to the topic it will listen to. To define such a mapping, we need:

• the topic name - we should receive it from above,

• the Kafka consumer name - we define it ourselves, analogously to the names of other artifacts,

• having both names, we can submit a request to the administrators to add a topic-to-Kafka-consumer mapping to the configuration (providing both names). Without the mapping, we will be able to write the Kafka consumer, but it will not receive messages.

Kafka Consumers in a Process

Kafka consumers can operate independently, but they are usually linked to a process in Eximee BPMS. Kafka consumers in a process participate in events of the "Message" type. At a step of this type, Eximee BPMS waits for a message so it can continue the process execution. The message can be sent from a Kafka consumer.

Figure 1. Example process with a step of type "Message Intermediate Catch Event" (event "Wait for notification from the bank")

During the process design stage, we must remember to fill in the message name in the configuration. It must meet the Eximee BPMS limitations (Message Events) and we will need it later when linking the process step to the Kafka consumer.

Figure 2. Example configuration of a step of type "Message Intermediate Catch Event"

Creating a Kafka Consumer

In Eximee Designer, we select appropriately Library → Kafka Consumers. Here, like with the other artifacts, there is a button Add Kafka Consumer:

Figure 3. Tab with artifact type: Kafka Consumers

Kafka consumers have limited access to the API and can use only the functions available in api.process.*. More information about the API: Process data operations and access

Example of using a Kafka consumer:

function callService(context) {
    // retrieve message content from the context
    var message = context.getMessage().value;
    // process the message
    var jsonValue = JSON.parse(stringValue);
    var processInstanceId = jsonValue.processInstanceId;
    // send a message to Eximee BPMS with information to continue the process
    api.process.v1.byInstanceId(messageValue.processInstanceId).correlateMessage('get-notification', null);
    return;
}

The message format is not imposed, so the structure and type of data it contains should be defined upfront in order to be able to process it correctly.

When writing a Kafka consumer, we will almost always start by retrieving the message content using context.getMessage().value;. How we can process the message later depends on the defined message format, so the set and type of fields may be different in each case. It is important to remember the process instance identifier during the agreement stage. In the example, this is the processInstanceId field. Thanks to this value, we know which process instance the given message concerns.

Retrieving data from a message

Using context.getMessage() we retrieve the message and its metadata. The object returned by this function has the following fields:

Name	Type	Description
value	string	The message content (payload).
headers	object	Kafka message headers as a key-value map
key	string	Kafka message key. It can be null if not provided.
topic	string	The name of the Kafka topic from which the message originates
partition	number	The topic partition number from which the message originates
offset	number	Timestamp (epoch ms) of message creation
timestamp	number	Message creation time

It should be noted that:

• context.getMessage() - retrieves the entire message object along with metadata,

• context.getMessage().value - retrieves only the message content. We will most often use this form.