Chapter 21. Kafka messages in a business process Red Hat AMQ Streams, based on Apache Kafka, is a streaming platform. It acts as a message broker, passing messages, which are sorted into topics, between applications in a software environment. Using Red Hat Process Automation Manager, you can create business processes that send and receive Kafka messages. You can use the following methods to create business processes that send and receive Kafka messages: Create a start event, intermediate catch event, or boundary event (attached to a human task) of the message type. KIE Server automatically subscribes to the Kafka topic that is defined in the message. A message triggers the event. The event node acts as the consumer of the message and can pass the content of the message to the subsequent node in the process. Create an end event or intermediate throw event of the type message . When the process triggers the event, KIE Server sends a Kafka message in the topic that is defined in the message. The message contains the data that is configured in the event. The event node acts as the producer of the message. Add the KafkaPublishMessages custom task to the process. This task does not require the KIE Server Kafka capability but can be more complicated to configure than message events. Configure your service and KIE Server with an emitter to automatically send Kafka messages about every completed process, case, and task when transactions are committed. 21.1. Creating an event that receives Kafka messages When designing your business process in Business Central, you can create an event that receives Kafka messages. This event is triggered each time a message arrives in the configured topic. The message must contain data that matches a predefined data object. The process engine parses the message and provides it as an output of the event. Prerequisites You are logged in to Business Central and have permission to edit business processes. You created a business process. Procedure In Business Central, open the project that contains your business process. Create a data object defining the data that the message must contain. For instructions about creating data objects, see Designing business processes using BPMN models . Select the business process and open the business process designer. Add a start event, an intermediate catch event, or a boundary event that is attached to a human task. The event must have the message type. Open the properties of the event. In the Message field, select New and then enter the name of the message. This name must be the same as the name of the topic from which the event is to receive Kafka messages, or else must be defined as topic-name in an org.kie.server.jbpm-kafka.ext.topics. topic-name KIE Server system property. For instructions about using org.kie.server.jbpm-kafka.ext.topics.* system properties to define topic names, see Chapter 22, Configuring KIE Server to send and receive Kafka messages from the process . Add an output data item and set the type of this data item to the data object that you created. Save the business process. Note If a Kafka message event starts a new process instance, the initiator field of the instance is set to unknown by default, because the Kafka consumer extension does not receive the identity of the user associated with the Kafka message. steps To enable Red Hat AMQ Streams integration when running the process, you must configure KIE Server. For more information about configuring KIE Server for Red Hat AMQ Streams integration, see Chapter 22, Configuring KIE Server to send and receive Kafka messages from the process . 21.2. Creating an event that sends Kafka messages When designing your business process in Business Central, you can create an event that sends Kafka messages. The event can have a data object as an input data item. The process engine sends the content of a data object as a message in the configured topic. Prerequisites You are logged in to Business Central and have permission to edit business processes. You created a business process. Procedure In Business Central, open the project that contains your business process. Create a data object defining the data that the message must contain. For instructions about creating data objects, see Designing business processes using BPMN models . Select the business process and open the business process designer. Add an intermediate throw event or an end event of the type message . Open the properties of the event. In the Message field, select New and then enter the name of the message. This name must be the same as the name of the topic to which the event is to send Kafka messages, or else must be defined as topic-name in an org.kie.server.jbpm-kafka.ext.topics. topic-name KIE Server system property. For instructions about using org.kie.server.jbpm-kafka.ext.topics.* system properties to define topic names, see Chapter 22, Configuring KIE Server to send and receive Kafka messages from the process . Add an input data item and set the type of this data item to the data object that you created. Save the business process. steps To enable Red Hat AMQ Streams integration when running the process, you must configure KIE Server. For more information about configuring KIE Server for Red Hat AMQ Streams integration, see Chapter 22, Configuring KIE Server to send and receive Kafka messages from the process . 21.3. Adding a custom task that sends Kafka messages You can add a KafkaPublishMessages custom task to your process. This task sends Kafka messages. It does not use the KIE Server Kafka capability, so you can use this task in processes that do not run on a KIE Server. However, this task can be more complicated to configure than other Red Hat AMQ Streams integration options. Prerequisites You are logged in to Business Central as an administrative user. Procedure In the Business Central administrative settings menu, select Custom Tasks Administration . Ensure that KafkaPublishMessages is set to On . In Business Central, select Menu Design Projects and then click the space name and the project name. Select the Settings Custom Tasks tab. In the KafkaPublishMessages line, click Install . Optional: Enter the following information. If you leave any of the fields empty, the process engine uses the default values for these fields. Bootstrap Servers : The host and port of the Kafka broker. You can use a comma-separated list of multiple host:port pairs. The default value is localhost:9092 . Client ID : An identifier string to pass to the broker when making requests. Red Hat AMQ Streams uses this string for logging. The default value is Key Serializer class : The class that provides the key serializer. The default value is org.apache.kafka.common.serialization.StringSerializer . You can enter a different value if you want to use your own custom serializer class. Value Serializer class : The class that provides the value serializer. The default value is org.apache.kafka.common.serialization.StringSerializer . You can enter a different value if you want to use your own custom serializer class. In any of these fields, you can enter an env[ property ] value. In this case, the process engine reads the setting from a system property at runtime. For example, you can set Client ID to env[application.client.id] and then, before running the process service, set the client ID value in the application.client.id system property. If you entered the name of a custom serializer class, complete the following additional configuration: Select the Settings Deployments Work Item Handlers tab. In the KafkaPublishMessages line, modify the Value field to add the classLoader parameter. For example, the initial value of this field can be the following string: In this example, change the value to the following string: Select the Assets tab. Select the business process and open the business process designer. Add the KafkaPublishMessages custom task, available under Custom Tasks in the BPMN modeler palette. In the properties of the custom task, open the data assignments. Assign the Key , Topic , and Value inputs to define the message. steps If you entered a custom serializer class, you must provide this class to your business application. For instructions about providing custom classes to your business application, see Chapter 24, Providing a custom class to your business application in Business Central .

Commands:
new org.jbpm.process.workitem.kafka.KafkaWorkItemHandler("127.0.0.1:9092", "jbpm", "com.myspace.test.MyCustomSerializer", "com.myspace.test.MyCustomSerializer")
new org.jbpm.process.workitem.kafka.KafkaWorkItemHandler("127.0.0.1:9092", "jbpm", "com.myspace.test.MyCustomSerializer", "com.myspace.test.MyCustomSerializer", classLoader)