A quickstart project that deals with traveller processing carried by rules. It illustrates how easy it is to make the Kogito processes and rules to work with Apache Kafka
This example shows
- consuming events from a Kafka topic and for each event start new process instance
- each process instance is expecting a traveller information in JSON format
- traveller is then processed by rules and based on the outcome of the processing (processed or not) traveller is
- if successfully processed traveller information is logged and then updated information is send to another Kafka topic
- if not processed traveller info is logged and then process instance finishes without sending reply to Kafka topic
- Diagram Properties (top)
- Diagram Properties (bottom)
- Diagram Properties (process variables)
- Start Message
- Start Message (Assignments)
- Process Traveler Business Rule (top)
- Process Traveler Business Rule (bottom)
- Process Traveler Business Rule (Assignments)
- Process Traveler Gateway
- Process Traveler Gateway Yes Connector
- Process Traveler Gateway No Connector
- Log Traveler Script Task
- Skip Traveler Script Task
- Processed Traveler End Message
- Processed Traveler End Message (Assignments)
- Skip Traveler End
This quickstart requires an Apache Kafka to be available and by default expects it to be on default port and localhost.
- Install and Startup Kafka Server / Zookeeper
https://kafka.apache.org/quickstart
You will need:
- Java 11+ installed
- Environment variable JAVA_HOME set accordingly
- Maven 3.8.6+ installed
mvn clean compile spring-boot:run
mvn clean package
To run the generated native executable, generated in target/
, execute
java -jar target/process-kafka-quickstart-springboot.jar
You can take a look at the OpenAPI definition - automatically generated and included in this service - to determine all available operations exposed by this service. For easy readability you can visualize the OpenAPI definition file using a UI tool like for example available Swagger UI.
In addition, various clients to interact with this service can be easily generated using this OpenAPI definition.
To make use of this application it is as simple as putting a message on travellers
topic with following content
- To examine ProcessedTravellers topic and verify upcoming messages will be processed
Execute in a separate terminal session
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic processedtravellers
- Send message that should be processed to Topic
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic travellers
Content (cloud event format)
{
"specversion": "0.3",
"id": "21627e26-31eb-43e7-8343-92a696fd96b1",
"source": "",
"type": "travellers",
"time": "2022-02-24T13:25:16+0000",
"data": {
"firstName" : "Jan",
"lastName" : "Kowalski",
"email" : "[email protected]",
"nationality" : "Polish"
}
}
One liner
{"specversion": "0.3","id": "21627e26-31eb-43e7-8343-92a696fd96b1","source": "","type": "travellers", "time": "2022-02-24T13:25:16+0000","data": { "firstName" : "Jan", "lastName" : "Kowalski", "email" : "[email protected]", "nationality" : "Polish"}}
this will then trigger the successful processing of the traveller and put another message on processedtravellers
topic.
To take the other path of the process put following message on travellers
topic
- Send Message to Topic
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic travellers
With the following content (Cloud Event Format)
{
"specversion": "0.3",
"id": "31627e26-31eb-43e7-8343-92a696fd96b1",
"source": "",
"type": "travellers",
"time": "2022-02-24T13:25:16+0000",
"data": {
"firstName" : "John",
"lastName" : "Doe",
"email" : "[email protected]",
"nationality" : "American"
}
}
One Liner
{"specversion": "0.3","id": "31627e26-31eb-43e7-8343-92a696fd96b1","source": "","type": "travellers", "time": "2022-02-24T13:25:16+0000","data": { "firstName" : "John", "lastName" : "Doe", "email" : "[email protected]", "nationality" : "American"}}
this will not result in message being send to processedtravelers
topic.
In the operator
directory you'll find the custom resources needed to deploy this example on OpenShift with the Kogito Operator.