I have posted articles here on how to use the ruleengine from Java code, but also with Apache Hadoop mapreduce, Pentaho Data Integration (ETL) and Apache Nifi - a Dataflow Management system.
The advantage is always the same: You don't hardcode business logic in your code or process. And that keeps your code cleaner and thus more agile. Agile, clean code directly translates into more quality and satisfaction for both you and your customers and it is also positively impacting time and cost.
Finally I found some time to also write about integrating the ruleengine into an Apache Kafka Consumer. The use case here is to check and validate your data - comming from a Kafka topic - before you further process it. But the ruleengine can not only validate the data, it can also apply actions to modify the data.
I have used a couple of times in the past the "Rent a Ferrari" scenario. Somebody wants to rent a ferrari. But there are some business rules to follow before the applicant can drive this shiny car. The rules are:
- Driver must have a valid drivers license
- Driver must be of age 26 or above
- If driver is less than 26 years old, he/she has to pay an extra fee
I went ahead and have setup a Kafka Connector to read data from a MySQL database. The relevant table is labeled "ferrari applications". The connector picks up changes to or additions of records based on the (autoincrement) id or based on the modified column, which is a timestamp.
One could imagine that the data for this table is collected in a web application of the car rental company. (Yes of course, the data could also go directly to a Kafka Topic from the application.)
This is what the table looks like:
As data arrives or is modified in the table the Kafka connector will pick it up and send it to a Kafka Topic "test_jdbc_ferrari_applications". The data arrives in Kafka as JSON like this:
{"id":7965,"name":"stefan","age":24,"drivers_license":1,"pays_extra_fee":1,"modified":1504342435000,"status":null}
{"id":7966,"name":"maria","age":28,"drivers_license":0,"pays_extra_fee":1,"modified":1504342435000,"status":null}
{"id":7967,"name":"maria","age":88,"drivers_license":1,"pays_extra_fee":1,"modified":1504342435000,"status":null}
{"id":7968,"name":"barbara","age":81,"drivers_license":1,"pays_extra_fee":0,"modified":1504342435000,"status":null}
{"id":7969,"name":"christian","age":52,"drivers_license":0,"pays_extra_fee":0,"modified":1504342435000,"status":null}
{"id":7970,"name":"nora","age":90,"drivers_license":1,"pays_extra_fee":1,"modified":1504342435000,"status":null}
{"id":7971,"name":"rudi","age":23,"drivers_license":0,"pays_extra_fee":0,"modified":1504342435000,"status":null}
{"id":7972,"name":"stefano","age":18,"drivers_license":0,"pays_extra_fee":0,"modified":1504342435000,"status":null}
The last part of the puzzle is a Java class that connects to Kafka and subscribes to the topic. When a record arrives, this record runs through the ruleengine. The business rules are run against the data. Based on if the business logic passed or failed, the relevant action - as described further above - is fired and the record is updated.
Here is a snippet from the Java code:
while (true)
{
// poll records from kafka
ConsumerRecords<String, String> records = kafkaConsumer.poll(numberOfRecordsToPoll);
// loop over the records
for (ConsumerRecord<String, String> record : records)
{
// create a collection of fields from the incomming record
RowFieldCollection collection = getRowFieldCollection(record.value());
// as we are in a loop, we remove the results of the ruleengine before we run it again
bre.getRuleExecutionCollection().clear();
// throw exception but continue in case of problems
// could be changes in the kafka record (missing fields) or missing data
try
{
// run the ruleengine using the record key and the collection of fields
// note that the business logic might contain actions that modify the data
bre.run(record.key(), collection);
....
....
Once run, the Java class waits for incomming records from Kafka and processes them. For this simple example, the validated and updated record in then output.
Here is the ouput:
record: [13112, lars, 45, 0, 1, 1504344186000, failed]
record: [13113, heinz, 20, 1, 0, 1504344186000, failed]
record: [13114, carmela, 28, 1, 0, 1504344186000, passed]
record: [13115, dieter, 41, 1, 1, 1504344186000, passed]
record: [13116, annie, 97, 1, 0, 1504344186000, passed]
record: [13117, anne, 30, 1, 0, 1504344186000, passed]
record: [13118, bruno, 47, 1, 0, 1504344186000, passed]
record: [13119, stefano, 15, 1, 1, 1504344186000, passed]
record: [13120, carmela, 37, 0, 0, 1504344186000, failed]
record: [13121, heinz, 24, 1, 0, 1504344186000, failed]
The data correponds to the data captured in the MySQL table, send to Kafka using Kafka connect and then picked up by the Java consumer class. The last column though is containing the results of the validation done with the business rules and the resulting action that was fired. The "status" column was updated to "failed" or "passed".
That's it. You could of course do further processing and e.g. store the data in another Kafka topic. A (micro-)service could pick it up from there and react on the results of the validation. Or you could make the results available for analytical purposes.
This is a simple example with simple business logic. But at this point you can modify the business logic and add additional rules and actions and you don't have to change the process or code. This could be done by a business expert reacting to changing conditions or contracts.
As you can see, the ruleengine plugs in everywhere. You not only get rid of the business logic in your code, but you also have a central place for your business logic. Again this is a plus for quality as also the maintenance of the logic is in one central place. Defining their logic using a central, user-oriented tool is also more transparent for the business users. And at last, when your colleagues leave the company you not anymore have to look for (ever-changing) business logic in their code and processes, spread all over the place.
Please try the ruleengine and send me your feedback.
Carpe Diem