Event processing consists of taking one or more events from an event stream and applying actions over those events. In general, in an enterprise service bus, there are commodity services; the most common are the following:

• Data transformation
• Event handling
• Protocol conversion
• Data mapping

Message processing in the majority of cases involves the following:

• Message structure validation against a message schema
• Given an event stream, filtering the messages from the stream
• Message enrichment with additional data
• Message aggregation (composition) from two or more message to produce a new message

This chapter is about event validation. The chapters that follow are about composition and enrichment.

The first step in event modeling is to express the event in English in the following form:

Subject-verb-direct object

For this example, we are modeling the event customer consults the ETH price:

• The subject in this sentence is customer, a noun in nominative case. The subject is the entity performing the action.
• The verb in this sentence is consults; it describes the action performed by the subject.
• The direct object in this sentence is ETH price. The object is the entity in which the action is being done.

We can represent our message in several message formats (covered in other sections of this book):

• JavaScript Object Notation (JSON)
• Apache Avro
• Apache Thrift
• Protocol Buffers

JSON is easily read and written by both humans and machines. For example, we could chose binary as the representation, but it has a rigid format and it was not designed for humans to...

This time, we are going to build our project with Gradle. The first step is to download and install Gradle from http://www.gradle.org/downloads.

Gradle only requires a Java JDK (version 7 or higher).

macOS users can install Gradle with the brew command, as follows:

$ brew update && brew install gradle

The output is something like the following:

==> Downloading https://services.gradle.org/distributions/gradle-4.10.2-all.zip
==> Downloading from https://downloads.gradle.org/distributions/gradle-4.10.2-al
######################################################################## 100.0%
  /usr/local/Cellar/gradle/4.10.2: 203 files, 83.7MB, built in 59 seconds

Linux users can install Gradle with the apt-get command, as follows:

$ apt-get install gradle

Unix users can install with sdkman, a tool for managing parallel versions of most Unix-based systems...

Now that we have our project skeleton, let's recall the project requirements for the stream processing engine. Remember that our event customer consults ETH price occurs outside Monedero and that these messages may not be well formed, that is, they may have defects. The first step in our pipeline is to validate that the input events have the correct data and the correct structure. Our project will be called ProcessingEngine.

The ProcessingEngine specification shall create a pipeline application that does the following:

• Reads each message from a Kafka topic called input-messages
• Validates each message, sending any invalid event to a specific Kafka topic called invalid-messages
• Writes the correct messages in a Kafka topic called valid-messages

These steps are detailed in Figure 2.1, the first sketch for the pipeline processing engine:

Our Reader invokes the process() method; this method belonging to the Producer class. As with the consumer interface, the producer interface encapsulates all of the common behavior of the Kafka producers. The two producers in this chapter implement this producer interface.

In a file called Producer.java, located in the src/main/java/monedero directory, copy the content of Listing 2.6:

package monedero;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
public interface Producer {
  void process(String message);                                 //1
  static void write(KafkaProducer<String, String> producer,
                    String topic, String message) {             //2
    ProducerRecord<String, String> pr = new ProducerRecord<>(topic, message);
    producer...

The ProcessingEngine class coordinates the Reader and Writer classes. It contains the main method to coordinate them. Create a new file called ProcessingEngine.java in the src/main/java/monedero/ directory and copy therein the code in Listing 2.8.

The following is the content of Listing 2.8, ProcessingEngine.java:

package monedero;
public class ProcessingEngine {
  public static void main(String[] args) {
    String servers = args[0];
    String groupId = args[1];
    String sourceTopic = args[2];
    String targetTopic = args[3];
    Reader reader = new Reader(servers, groupId, sourceTopic);
    Writer writer = new Writer(servers, targetTopic);
    reader.run(writer);
  }
}

ProcessingEngine receives four arguments from the command line:

• args[0] servers, the host and port of the Kafka broker
• args[1] groupId, the consumer group...

The Writer class implements the producer interface. The idea is to modify that Writer and build a validation class with minimum effort. The Validator process is as follows:

• Read the Kafka messages from the input-messages topic
• Validate the messages, sending defective messages to the invalid-messages topic
• Write the well-formed messages to valid-messages topic

At the moment, for this example, the definition of a valid message is a message t0 which the following applies:

• It is in JSON format
• It contains the four required fields: event, customer, currency, and timestamp

If these conditions are not met, a new error message in JSON format is generated, sending it to the invalid-messages Kafka topic. The schema of this error message is very simple:

{"error": "Failure description" }

The first step is create a new Validator.java file...

At the moment, the ProcessingEngine class coordinates the Reader and Writer classes. It contains the main method to coordinate them. We have to edit the ProcessingEngine class located in the src/main/java/monedero/ directory and change Writer with Validator, as in Listing 2.10.

The following is the content of Listing 2.10, ProcessingEngine.java:

package monedero;
public class ProcessingEngine {
  public static void main(String[] args) {
    String servers = args[0];
    String groupId = args[1];
    String inputTopic = args[2];
    String validTopic = args[3];
    String invalidTopic = args[4];
    Reader reader = new Reader(servers, groupId, inputTopic);
    Validator validator = new Validator(servers, validTopic, invalidTopic);
    reader.run(validator);
  }
}

ProcessingEngine receives five arguments from the command line:

• args...

In this chapter we learned how to model the messages in JSON format and how to set up a Kafka project with Gradle.

Also, we learned how to write to and read from Kafka with a Java client, how to run the processing engine, how to code a validator in Java, and how to run the message validation.

In Chapter 3, Message Enrichment, the architecture of this chapter will be redesigned to incorporate message enrichment.