Kafka consumers

Availability of the 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 for document verification, but this takes time, 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 process execution.

Kafka topic binding to a Kafka Consumer

Each Kafka consumer must be bound to the topic it will listen to. To establish such a mapping, we need:

• the topic name - we should receive it from above,
• the Kafka consumer name - we determine it ourselves, similarly to the names of other artifacts,
• with both names, we can submit a request to 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.

Configuration requirements on the Kafka side

For each topic used by a low-code consumer, administrators must prepare not only the main topic, but also a complete set of retry and dead-letter topics:

• <topic-name>-retry-0
• <topic-name>-retry-1
• <topic-name>-dlt

If any of these topics does not exist, the configuration is incomplete and the retry and DLT handling mechanism will not work correctly.

This requirement applies to absolutely every topic handled by the low-code consumer.

Additionally, the Eximee consumer must be granted the permission idempotentWrite, because as part of error handling it writes messages to retry and DLT topics.

Example of a complete topic configuration

If the low-code consumer uses the topic eximee-mortgage.fct.email-status, prepare the whole set:

• eximee-mortgage.fct.email-status
• eximee-mortgage.fct.email-status-retry-0
• eximee-mortgage.fct.email-status-retry-1
• eximee-mortgage.fct.email-status-dlt

When submitting the request to administrators, also provide information that the Eximee consumer needs the permission idempotentWrite.

Error handling and message redelivery

The low-code consumer can finish message handling in two ways:

1. It does not throw an exception - the message is considered successfully processed.
2. It throws an exception MessageNotReadyToProcessException and forces a retry.

By default, the low-code consumer uses the following retry configuration:

• maxAttempts = 3
• initialInterval = 60000 ms
• multiplier = 5.0
• maxInterval = 300000 ms

This means that, by default, 3 attempts are made in total:

1. the first attempt on the main topic,
2. the second attempt after 1 minute on the topic -retry-0,
3. the third attempt after 5 minutes on the topic -retry-1.

If the third attempt also ends with an exception, the message goes to the topic -dlt.

The error-handling flow is therefore as follows:

1. The message goes to the main topic.
2. If the consumer throws an exception, after 1 minute the message goes to the topic with the suffix -retry-0.
3. If the second attempt also ends with an exception, after 5 minutes the message goes to the topic with the suffix -retry-1.
4. If the third attempt also ends with an exception, the message goes to the topic with the suffix -dlt.

Configuring custom retry times

Retry times are configured per kafkaId using the properties:

• script.events.<kafkaId>.retry.max-attempts
• script.events.<kafkaId>.retry.initial-interval
• script.events.<kafkaId>.retry.multiplier
• script.events.<kafkaId>.retry.max-interval

Configuration example:

script:
  events:
    centralKafka:
      retry:
        max-attempts: 4
        initial-interval: 30000
        multiplier: 2.0
        max-interval: 180000

In this example:

• after the first error, the next attempt will occur after 30 seconds,
• the next delays will be calculated using the multiplier 2.0,
• a single delay will not exceed 180 seconds,
• the total number of attempts will be 4.

After the message is moved to the Dead Letter Topic, the message is no longer automatically forwarded or reprocessed. Administrator intervention is required, and they should analyze the cause of the error and decide on further actions.

The presence of the DLT topic does not solve the problem automatically. In the current implementation there is no dedicated "do not try anymore" exception, and there is no automatic further processing of messages from the DLT. This is the place where messages requiring manual analysis and administrative intervention are sent.

Kafka Consumers in the Process

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

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

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

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

Creating a Kafka Consumer

In Eximee Designer, select Library → Kafka Consumers. Here, just like for 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 operations and data access

Example of using a Kafka consumer:

function callService(context) {
    // get the message content from the context
    let message = context.getMessage().value;
    // processing the message
    let jsonValue = JSON.parse(stringValue);
    let 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 enforced, so the structure and type of the data contained in it should be determined in advance so that it can be processed correctly.

When writing a Kafka consumer, we will almost always start by retrieving the message content using context.getMessage().value;. How we can later process the message depends on the established message format, so the set and type of fields may be different in each case. It is important to remember the process instance identifier at the design 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	Message content (payload).
headers	object	Kafka message headers as a key–value map
key	string	Kafka message key. It can be null if it has not been filled in.
topic	string	Name of the Kafka topic from which the message comes
partition	number	Partition number of the topic from which the message comes
offset	number	Timestamp (epoch ms) of message creation
timestamp	number	Message creation time

Please note that:

• context.getMessage() - retrieves the entire message object along with metadata,
• context.getMessage().value - retrieves only the message content. Most often we will use this form.