Chapter 9. Commands

So far, we have concerned ourselves almost exclusively with events. Events, as we explained in chapter 1, are discrete occurrences that take place at a specific point in time. Throughout this book, we have created events, written them to our unified log, read them from our unified log, validated them, enriched them, and more. There is little that we have not done with events!

But we can represent another unit of work in our unified log: the command. A command is an order or instruction for a specific action to be performed in the future—each command, when executed, produces an event. In this chapter, we will show how representing commands explicitly as one or more streams within our unified log is a powerful processing pattern.

We will start with a simple definition for commands, before moving on to show how decision-making apps can operate on events in our unified log and express decisions in the form of commands. Modeling commands for maximum flexibility and utility is something of an art; we will cover this next before launching into the chapter’s applied example.

Working for Plum, our fictitious global consumer-electronics manufacturer, we will define a command intended to alert Plum maintenance engineers of overheating machines on the factory floor. We will represent this command in the Apache Avro schema language and use the Kafka producer script from chapter 2 to publish those alert commands to our unified log.

We will then write a simple command-executor app by using the Kafka Java client’s producer and consumer capabilities. Our command executor will read the Kafka topic containing our alert commands and send an email to the appropriate support engineer for each alert; we will use Rackspace’s Mailgun transactional email service to send the emails.

Finally, we will wrap up the chapter with some design questions to consider before adding commands into your company’s own unified log. We will consider the pros and cons of two designs for your command topic(s) in Kafka or Kinesis and will also introduce the idea of a hierarchy of commands.

Let’s get started!

9.1. Commands and the unified log

What is a command exactly, and what does it mean in the context of the unified log? In this section, we’ll introduce commands as complementary to events, and argue for making decision-making apps, and the commands they generate, a central part of your unified log.

9.1.1. Events and commands

A command is an order or instruction for a specific action to be performed in the future. Here are some example commands:

If an event is a record of an occurrence in the past, then a command is the expression of intent that a new event will occur in the future:

Figure 9.1 illustrates the behavioral flow underpinning the first example: a decision (I want pizza) produces a command (“Order a pizza, Dad”), and if or when that command is executed, then we can record an event as having taken place (my dad ordered a pizza). In grammatical terms: an event is a past-tense verb in the indicative mood, whereas a command is a verb in the imperative mood.

You can see that there is a symbiotic relationship between commands and events: strictly speaking, without commands, we would have no events.

9.1.2. Implicit vs. explicit commands

If behind every event is a command, how could we have reached chapter 9 in a book all about events without encountering a single command? The answer is that almost all software relies on implicit commands—the decision to do something is immediately followed by the execution of that decision, all in the same block of code. Here is a simple example of this in pseudocode:

This code includes no explicit commands: instead, we have a block of code that mixes decision-making in with the execution of those decisions. How would this code look if we had explicit commands? Something like this:

Of course, this version is not functionally identical to the previous code block. In this version, we are only returning commands as the result of our decisions; we are not actually executing those commands. There will have to be another piece of code downstream of make_decision(obj), looking something like this:

It looks like we have turned something simple—making a decision and acting on it—into something more complicated: making a decision, emitting a command, and then executing that command. The reason we do this is in support of separation of concerns.^[1] Turning our commands into explicit, first-class entities brings major advantages:

¹You can learn more about separation of concerns at Wikipedia: https://en.wikipedia.org/wiki/Separation_of_concerns.

This decoupling of decision-making and command execution clearly has benefits, but if we involve our unified log, things get even more powerful. Let’s explore this next.

9.1.3. Working with commands in a unified log

In the previous section, we split a single function, decide_and_act(obj), into two separate functions:

With a unified log, there is no reason that these two functions need to live in the same application: we can create a new stream called commands and write the output of make_decision(obj) into that stream. We can then write a separate stream-processing job that consumes the commands stream and executes the commands found within by using execute_command(cmd). This is visualized in figure 9.2.

With this architecture, we have completely decoupled the decision-making process from the execution of the commands. Our stream of commands is now a first-class entity: we can attach whatever applications we want to execute (or even just observe) those commands.

In the next section, we will add commands to our unified log.

9.2. Making decisions

To generate commands, we’ll first have to make some decisions. In this section, we’ll continue working with Plum, our fictional company with a unified log introduced in chapter 7, and wire some decision-making into that unified log. Let’s get started.

9.2.1. Introducing commands at Plum

Let’s imagine again that we are in the BI team at Plum, a global consumer-electronics manufacturer. Plum is in the process of implementing a unified log; for unimportant reasons, the unified log is a hybrid, using Amazon Kinesis for certain streams and Apache Kafka elsewhere. In reality, this is a fairly unusual setup, but it enables us to work with both Kinesis and Kafka in this section of the book.

At the heart of the Plum production line is the NCX-10 machine, which stamps new laptops out of a single block of steel. Plum has 1,000 of these machines in each of its 10 factories. Each machine emits key metrics as a form of health check to a Kafka topic every 5 minutes, as shown in figure 9.3.

We have been asked by the plant maintenance team at Plum to use the health-check data to help keep the NCX-10 machines humming. The team members are particularly concerned about one scenario that they want to tackle first: if a machine shows signs of overheating, a maintenance engineer needs to be alerted immediately in order to be able to inspect the machine and fix any issues with the cooling system.

We can deliver what plant maintenance wants by writing two stream-processing jobs:

Figure 9.4 depicts these two jobs and the Kafka streams between them.

Does the decision-making job sound familiar? It’s likely a piece of either single or multiple (stateful) event processing, as we explored in part 1. Because this chapter is about commands, we’ll skip over the implementation of the decision-making job and head straight to defining the command itself.

9.2.2. Modeling commands

Our decision-making job needs to emit a command to alert a maintenance engineer that a specific NCX-10 machine shows signs of overheating. It’s important that we model this command correctly: it will represent the contract between the decision-making job, which will emit the command, and the command-executing job, which will read the command.

We know what this command needs to do—alert a maintenance engineer—but how do we decide what fields to put in our command? Like a tasty muffin, a good command should have these characteristics:

Here is an example of a badly designed self-describing alert command for Plum:

Let’s consider how the executor would have to be written for this command, in pseudocode:

Unfortunately, this breaks both of the rules for a good command:

Here is a better version of an alert command for Plum:

The command may be a little more verbose, but the executor implementation for this command is much simpler:

This command requires the executor to do much less work: the executor now only has to send the email to the recipient and record an event to the same effect. The executor does not need to know that this specific alert relates to an overheating machine in a factory; the specifics of this alert were “fully baked” upstream, in the business logic of the decision-making job. This strong decoupling has the major advantage of making our executor general-purpose. Plum can write other decision-making jobs that can reuse this command-executing job without requiring any code changes in the executor.

The other thing to note is that we have made the alert command definitive but not overly restrictive. If we wanted to, we could update our executor to include a prioritization system for processing alerts, like so:

Using our unified log for commands makes updates like this achievable:

9.2.3. Writing our alert schema

We are now ready to define a schema for our alert command. We are going to use Apache Avro (https://avro.apache.org) to model this schema. As explained in chapter 6, Avro schemas can be defined in plain JSON files. The schema file needs to live somewhere, so we’ll add it into our command-executor app. We’ll need to create this app next.

We are going to write the command executor as a Java application, called ExecutorApp, using Gradle. Let’s get started. First, create a directory called plum. Then switch to that directory and run this:

Gradle has created a skeleton project in that directory, containing a couple of Java source files for stub classes, called Library.java and LibraryTest.java. Delete these two files, as we will be writing our own code shortly.

Next let’s prepare our Gradle project build file. Edit the file build.gradle and replace its current contents with the following listing.

Let’s check that we can build this:

Now we are ready to work on the schema for our new command.

9.2.4. Defining our alert schema

To add our new schema for alerts to the project, create a file at this path:

Populate this file with the Avro JSON schema in the following listing.

Quite a lot is going on in this Avro schema file. Let’s break it down:

A minor piece of housekeeping—we need to soft-link the resources/avro subfolder to another location so that the Avro plugin for Gradle can find it:

With our Avro schema defined, let’s use the Avro plugin in our build.gradle file to automatically generate Java bindings for our schema:

You will find the generated files inside your Gradle build folder:

These files are too lengthy to reproduce here, but if you open them in your text editor, you will see that the files contain POJOs to represent the three records and one enumeration that make up our alert.

This is a wrap for designing our alert command for Plum and modeling that command in Apache Avro. Next, we can proceed to building our command executor.

9.3. Consuming our commands

Imagine that Plum now has a Kafka topic containing a constant stream of our alert commands, all stored in Avro format. With this stream in place, Plum now needs to implement a command executor, which will consume those alerts and execute them. First, we will get all the plumbing in place and check that we can successfully deserialize the incoming Avro event.

9.3.1. The right tool for the job

We could use lots of different stream-processing frameworks to execute our commands, but remember that command execution involves only two tasks:

Figure 9.5 shows the specifics of these two tasks for Plum.

Both tasks are performed on one command at a time; there’s no need to consider multiple commands at once. Our command execution is analogous to the single-event processing we explored in chapter 2. As with single-event processing, we will need only a simple framework to execute our commands, so the simple consumer and producer capabilities of the Kafka Java client library from chapter 2 will fit the bill nicely.

9.3.2. Reading our commands

As a first step, we need to read in our individual commands as records from our Kafka topic commands. Remember from chapter 2 that this is called a consumer in Kafka parlance. As in that chapter, we will write our own consumer in Java, using the Kafka Java client library.

Let’s create a file for our consumer, called src/main/java/plum/Consumer.java, and add in the code in listing 9.3. This is a direct copy of the consumer code in chapter 2, except for the following changes:

So far, so good—we have defined a consumer that will read all the records from a given Kafka topic and hand them over to the execute method of the supplied executor. Next, we will implement an initial version of the executor. This won’t yet carry out the command, but it will show that we can successfully parse our Avro-structured events into Java objects.

9.3.3. Parsing our commands

See how our consumer is going to run the IExecutor.execute() method for each incoming command? To keep things flexible, the two executors we write in this chapter will both conform to the IExecutor interface, letting us easily swap out one for the other. Let’s now define this interface in another file, called src/main/java/plum/IExecutor.java. Add in the code in the following listing.

Again, this code is extremely similar to the IProducer we wrote in chapter 2. As with that IProducer, this interface contains static helper methods to configure a Kafka record producer and write events to Kafka. We need these capabilities in our executor so that we can fulfill the second part of executing any given command: emitting an event back into Kafka and recording this command as having been successfully executed.

With the interface in place, let’s write our first concrete implementation of IExecutor. At this stage, we won’t execute the command, but we want to check that we can successfully parse the incoming command. Add the code in the following listing into a new file called src/main/java/plum/EchoExecutor.java.

The EchoExecutor is simple. Every time the execute method is invoked with a serialized command, it does the following:

9.3.4. Stitching it all together

We can now stitch these three files together via a new ExecutorApp class containing our main method. Create a new file called src/main/java/plum/ExecutorApp.java and populate it with the contents of the following listing.

We will pass four arguments into our StreamApp on the command line:

Let’s build our stream processing app now. From the project root, the plum folder, run this:

We are now ready to test our stream processing app.

9.3.5. Testing

To test our new application, we are going to need five terminal windows. Figure 9.6 sets out what we’ll be running in each of these terminals.

Each of our first three terminal windows will run a shell script from inside our Kafka installation directory:

In our first terminal, we start up ZooKeeper:

In our second terminal, we start up Kafka:

In our third terminal, let’s start a script that lets us send commands into our alerts Kafka topic:

Let’s now give this producer an alert command, by pasting this into the same terminal, making sure to add a newline after the command to send it into the Kafka topic:

Phew! We are finally ready to start up our new command-executing application. In a fourth terminal, head back to your project root, the plum folder, and run this:

This has kicked off our app, which will now read all commands from alerts and execute them. Wait a second and you should see the following output in the same terminal:

Good news: our simple EchoExecutor is working a treat. Now we can move onto the more complex version to execute the alert and log the sent email. Shut down the stream processing app with Ctrl-Z and then type kill %%, but make sure to leave that terminal and the other terminal windows open for the next section.

9.4. Executing our commands

Now that we can successfully parse our alert commands from Avro-based JSONs into POJOs, we can now move on to executing those commands. In this section, we will wire in a third-party email service to send our alerts, and then finish up by emitting an event to record that the email has been sent. Let’s get started.

9.4.1. Signing up for Mailgun

Our overlords at Plum want the monitoring alerts about the NCX-10 machines to be sent to the maintenance engineers via email. So, we need to integrate some kind of email-sending mechanism into our executor. We have many to choose from out there, but we will go with a hosted transactional email service called Mailgun (www.mailgun.com).

Mailgun lets us sign up for a new account and start sending emails without providing any billing information, which is perfect for this kind of experimentation. If you prefer, you could equally use an alternative provider like Amazon Simple Email Service (SES), SendGrid, or even a local mail server option such as Postfix. If you go with an alternative email service provider, you will need to adjust the following instructions accordingly.

Head to the signup page for Mailgun: https://signup.mailgun.com/new/signup.

Fill in your details; you don’t need to provide any billing information. Click the Create Account button, and on the next screen you will see some Java code for sending an email, as shown in figure 9.7. Before you can send an email, you have to do two more things:

9.4.2. Completing our executor

With these steps completed, we are now ready to update our executor to send emails. First, we need a thin wrapper around the Mailgun email-sending functionality: create a file called src/main/java/plum/Emailer.java and add in the contents of the following listing.

Emailer.java contains a simple wrapper around the Mailgun emailing library that we are using. Make sure to update the constants with the API key and sandbox server details found in your Mailgun account.

With this in place, we can now write our full executor. Like the EchoExecutor, this will implement the IExecutor interface, but the FullExecutor will also send the email via Mailgun and log the event to our events topic in Kafka. Add the contents of the following listing into the file src/main/java/plum/FullExecutor.java.

FullExecutor is similar to our earlier EchoExecutor, but with two additions:

Now we need to rewrite the entry point to our app to use our new executor. Edit the file src/main/java/plum/ExecutorApp.java and repopulate it with the contents of the following listing.

Let’s rebuild our app now. From the project root, the plum folder, run this:

We are now ready to rerun our command executor.

9.4.3. Final testing

Head back to the terminal running the previous build of the executor. If it is still running, kill it with Ctrl-C. Now restart it:

Leave ZooKeeper and Kafka running in their respective terminals; we now need to start the fifth and final terminal, to tail our Kafka topic containing events:

Now let’s head back to the terminal that is running a producer connected to our alerts topic. We’ll take the alert command we ran before, update it so that the recipient’s email is the same one that we authorized with Mailgun, and then paste it into the producer:

For this to work, you must update the email in the alert to your own email, the one that you authorized with Mailgun. Check back in your new terminal that is tailing our Kafka events topic and you should see this:

That’s encouraging; it suggests that our command executor has sent our email. Check in your email client and you should see an incoming email, something like figure 9.8.

Great! We have now implemented a command executor that can take incoming commands—in our case, alerts for a long-suffering Plum maintenance engineer—and convert them into emails to that engineer. Our command executor even emits an email_sent event to track that the action has been performed.

9.5. Scaling up commands

This chapter has walked you through a simple example of executing an alert command as it passes through Plum’s unified log. So far, so good—but how do we scale this up to a real company with hundreds or thousands of possible commands? This section has some ideas.

9.5.1. One stream of commands, or many?

In this chapter, we called our stream of commands alerts on the basis that it would contain only alert commands, and our command-executor app processed all incoming commands as alerts. If we want to extend our implementation to support hundreds or thousands of commands, we have several options:

Figure 9.9 illustrates these three options.

When choosing one of these options, you will want to consider the operational overhead of having multiple streams versus the development overhead of making your commands self-describing. It is also important to understand how these different setups fare in the face of command-execution failures.

9.5.2. Handling command-execution failures

Imagine that Mailgun has an outage (whether planned or unplanned) for several hours. What will happen to our command executor? Remember that our command executor is performing single-event processing; it has no way of knowing that a systemic problem has arisen with Mailgun. If we are lucky, our Mailgun email-sending code will simply time out after perhaps 30 seconds, for each alert command that is processed. In this case, we would likely emit an Email Failed to Send event or similar back into Kafka.

Now imagine that our command executor is reading a stream containing many types of commands. The failure of each alert command costs us 30 seconds of processing time, so if our decision-making apps are generating new commands at a faster rate than this, our command executor will fall further and further behind. Other high-priority commands that Plum wants to execute will be extremely delayed.

Note that the sharded nature of a Kafka or Kinesis stream doesn’t help you: assuming that the failing command type is distributed across all of the shards, the worker assigned to each shard is equally likely to suffer slowdowns as the other workers. This is illustrated in figure 9.10.

Separating the commands into streams with different priorities does not necessarily help either: it is still possible for a high-priority command type to slow other high-priority commands, as also seen in figure 9.10.

Two potential solutions to consider are as follows:

9.5.3. Command hierarchies

In this chapter, we have argued in favor of creating commands that are shrink-wrapped and fully baked. A command executor should have discretion as to how each command should be best executed, but the executor should not have to look up additional data or contain command-specific business logic. Following this design, we created an alert command for Plum, which was resolved by our executor into an email sent via Mailgun.

The alert command is a useful one, but under the hood our command executor was resolving it into a send email command, and then ultimately into a send email via Mailgun command. For your unified log, you may want to model these more granular commands as well as the high-level ones, so that a decision-making app can be as precise or flexible as it likes as to the commands it emits. Figure 9.11 illustrates this idea of a form of command hierarchy.

Summary