microprofile-reactive-messaging-kafka: MicroProfile Reactive Messaging with Kafka QuickStart

This quickstart includes integration tests, which are located under the src/test/ directory. The integration tests verify that the quickstart runs correctly when deployed on the server.

Follow these steps to run the integration tests.

Copy across the microprofile-reactive-messaging-kafka/src/main/java/org/wildfly/quickstarts/microprofile/reactive/messaging/MockExternalAsyncResource.java file to your project. This class mocks a call to an asynchronous external resource. The details of how it is implemented are not important for this QuickStart.

MockExternalAsyncResource has one callable method:

The CompletionStage returned by this method will complete with a String when ready. A String is ready every two seconds. It will emit the following Strings in the given order:

After this initial sequence the returned CompletionStage will complete with a random entry from the above list. A new entry is available every two seconds.

Later we will wrap the strings in a TimedEntry object which contains the String and a timestamp. Since we will be storing it in a database, we add JPA annotations to it:

MicroProfile Reactive Messaging is based on CDI, so all interaction with the MicroProfile Reactive Messaging streams must happen from a CDI beans. Note: The beans must have either the @ApplicationScoped or @Dependent scopes.

Then within these beans we have a set of methods using the @Incoming and @Outgoing annotations from the MicroProfile Reactive Messaging specification to map the underlying streams. For an @Outgoing annotation its value specifies the Reactive Messaging stream to send data to, and for an @Incoming annotation its value specifies the Reactive Messaging stream to read data from. Although in this QuickStart we are putting all these methods into one CDI bean class, they could be spread over several beans.

The MicroProfile Reactive Messaging specification contains a full list of all the valid method signatures for such @Incoming/@Outging methods. We will use a few of them, and see how they make different use-cases easier.

Our bean looks as follows, and this is the main focus for the functionality in this QuickStart. We will also be using some other technologies, but they are secondary to this section. Explanations of each method will be given in line.

First we inject our mock asynchronous external data producer, which produces a string every two seconds. We explained this class in a previous section.

We inject a bean that will be used to persist entries to a RDBMS later on.

Now we get to the reactive messaging part.

Our first method is a 'producer' method, since is annotated with the @Outgoing annotation. It simply relays the output of our mock producer bean to the reactive messaging system. It uses the channel source (as indicated in the @Outgoing annotation’s value) as the target stream to send the data to. You can think of 'producer' methods as the entry point to push data into the reactive messaging streams.

Next we have a 'processor' method. It is annotated with both @Incoming and @Outgoing annotations so it gets data from the reactive messaging streams and then pushes it to another stream. Essentially it simply relays data.

In this case we get data from the source stream, so it will receive the entries made available by the sendInVm() method above, and forwards everything onto the filter stream.

In this case, since we just want to log the strings that were emitted, we are using a simple method signature receiving and returning the raw string provided. The Reactive Messaging implementation has unwrapped the CompletionStage from the previous method for us.

Note that there is no Kafka involved yet. Since the @Incoming and @Outgoing values match up, Reactive Messaging will use internal, in-memory streams.

Now we have another 'processor' method. We get the data from the filter stream (what was relayed by the previous logAllMessages() method) and forward it on to the sender stream.

In this method we want to do something a bit more advanced, namely apply a filter to the messages. We use a method receiving and returning a Reactive Stream, in this case we use PublisherBuilder from the MicroProfile Reactive Streams Operators specification. PublisherBuilder extends the Publisher interface from the Reactive Streams specification, and provides us with the filter() methods we will use here.

Again the Reactive Messaging implementation does all the wrapping for us.

Our filter method tells us to only relay messages that match Hello or Kafka, and drop everything else. In other words, later methods in the stream will only receive occurrences of Hello or Kafka.

Next we have another 'processor' method, which receives data from the sender stream and forwards it on to the to-kafka stream. It’s parameter is a simple String, MicroProfile Reactive Messaging will unwrap the stream from the PublisherBuilder returned in the previous method and call this next method with the individual entries.

In this method we want to wrap up the data into the TimedEntry class we defined earlier, so we have tuple of the message and a timestamp.

Additionally we want to set a Kafka key for the entries so that we can take advantage of Kafka’s querying capabilities (not done in this quickstart). In order to do this, we do the following steps:

Our final method is a 'consumer' method, as it has only an @Incoming annotation. You can think of this as a 'final destination' for the data in your application. We are using a Message<TimedEntry> as our method parameter. We are using this signature since we want to access the IncomingKafkaRecordMetadata, which contains the key we added in the previous method and additional information such as the Kafka partition and topic the message was sent on. Since we are using the signature taking a Message as the parameter, we need to ack() the message and return the resulting CompletionStage<Void>. (If we don’t want to ack the receipt of the message and are not interested in the IncomingKafkaRecordMetadata, we could have used a simpler signature such as void receiveFromKafka(TimedEntry message).)

The methid calls through to our dbBean to store the received data in a RDBMS. We will look at this briefly later.

You might have noticed that up to, and including, the sendToKafka() method the @Incoming.value() matches the @Outgoing.value() of the prior method. This indicates that these streams (source, filter and sender) are handled in memory by the Reactive Messaging implementation.

For the last two methods this is different and there is no such pairing. The sendToKafka() method sends its data to the to-kafka stream:

However, there are no methods annotated with @Incoming("to-kafka).

And the receiveFromKafka() method is expecting to receive data from the from-kafka stream:

Again, there are no methods annotated with @Outgoing("from-kafka").

These 'unpaired' sets of methods indicate that we do not want to use an in-memory stream, and want to use an external system for these streams. If we were try to deploy the MessagingBean in this state the application would fail to deploy. To fix this, and tell it what to map onto, we need to provide some configuration.

We have covered all the reactive messaging parts, but have missed out how the MessagingBean.receiveFromKafka() stores data via the DatabaseBean. This is not the focus of this QuickStart, so we will just mention how this works quickly.

This is the definition of DatabaseBean:

It injects an EntityManager for the persitence context test, which is defined in the src/main/resources/META-INF/persistence.xml file:

The DatabaseBean.store() method saves the TimedEntry and the DatabaseBean.loadAllTimedEntries() method loads all the ones we stored.

It is worth pointing out that the @Incoming and @Outgoing annotated methods called by the Reactive Messaging implementation (such as MessagingBean.receiveFromKafka()) happen outside of user space, so there is no @Transaction associated with them. So we need to annotated the DatabaseBean.store() method with @Transactional in order to save our entry to the database.

Finally, we would like to be able to view the data that was stored in the database. To do this we will add a JAX-RS endpoint that queries the database by calling DatabaseBean.loadAllTimedEntries().

We expose our JAX-RS application at the context path:

So far what we have seen is really happening in the back-end with little user interaction. The MicroProfile Reactive Messaging 2.0 specification adds a @Channel annotation and an Emitter interface which makes it easier to send data to MicroProfile Reactive Messaging streams from user initiated code and to receive data from Reactive Messaging streams.

To showcase this functionality we add another CDI bean called UserMessagingBean:

Looking at this in more detail, the following field

is used to send data to the MicroProfile Reactive Messagin stream called user, which is done in the following method

@Inject @Channel on an Emitter can be considered similiar to an @Outgoing annotated method but with the data coming from code paths invoked by user interaction. In this case we are not using Kafka to back the stream but if we wanted to, for this example, the MicroProfile Config properties would be prefixed with the mp.messaging.outgoing.user. prefix.

Next we have the constructor where we inject a Publisher (We could also have used a MicroProfile Reactive Streams Operators PublisherBuiilder) to define the receiving side.

We store this injected Publisher in the broadcastPublisher field. We will come back to why we are wrapping it in a BroadcastPublisher in a second. So an @Inject @Channel on a Publisher (or PublisherBuilder) can be considered equivalent to use of the @Incoming annotation. In this case we are listening to the user in memory stream so messages sent via the Emitter will be received on this Publisher. If instead we wanted to configure it to send via Kafka we would use MicroProfile Config properties prefixed with the mp.messaging.incoming.user. prefix.

There are a few caveats on this mechanism though:

The above two points will hopefully be fixed in a future version of the specification. For the purposes of this quickstart we are bypassing the above limitations by creating the BroadcastPublisher class and wrapping the original Publisher in that instead. Note that implementing Publisher is hard, and that BroadcastPublisher should be considered a 'proof of concept'. For more details about the BroadcastPublisher see the source code of the UserMessagingBean class. In a nutshell what it does is:

Finally we have a JAX-RS endpoint

It simply delegates the values received from POST requests under /user/<value> onto the bean which sends them via the Emitter.

Then GET requests for /user return the BroadCastPublisher to the user who will then receive data received on the MicroProfile Reactive Messaging channel.

Now you should be able to compile the application and prepare it for deployment.

We need to first start Apache Kafka. Then we will run the WildFly server and deploy our application to the server.

With the application still running, open two terminal windows. Enter the following curl command in both of them

The -N option keeps the connection open, so we receive data as it becomes available on the publisher.

In a third terminal window enter the commands:

In the first two terminal windows you should see these entries appear as they are posted:

The integration tests included with this quickstart, which verify that the quickstart runs correctly, may also be run with a bootable jar.

Follow these steps to run the integration tests.

On OpenShift, the S2I build with Apache Maven uses an openshift Maven profile to provision a WildFly server, deploy and run the quickstart in OpenShift environment.

The server provisioning functionality is provided by the WildFly Maven Plugin, and you may find its configuration in the quickstart pom.xml:

You may note that unlike the provisioned-server profile it uses the cloud feature pack which enables a configuration tuned for OpenShift environment.

This section contains the basic instructions to build and deploy this quickstart to WildFly for OpenShift or WildFly for OpenShift Online using Helm Charts.

The integration tests included with this quickstart, which verify that the quickstart runs correctly, may also be run with the quickstart running on OpenShift.

Run the integration tests using the following command to run the verify goal with the integration-testing profile activated and the proper URL: