Micronaut is a new JVM-based framework designed to make creating microservices quick and easy. One of the most exciting features of Micronaut is its support for reactive programming and we will see in this article how to store and search data in Elassandra in a reactive way through the Cassandra driver and run unit tests with JUnit5.
Here a basic JSON sales receipt with 3 products items:
In order to handle conversion between Cassandra types and custom Java object , and generate CQL queries, the java Cassandra driver provides a Cassandra mapper. By combining Jackson, Lombok and Cassandra mapper java annotations in the Basket POJO object, we get both JSON serialization and Cassandra Data Access Object.
In the same way, the BasketItem is mapped to a Cassandra User Defined Type as shown bellow. This will be a nested object in the Elasicsearch documents.
Finally, BascketStatus, a java enum, is managed through a registered codec as described in the driver documentation. Unfortunately, the java Cassandra mapper cannot generate the CQL schema, so we need to write it manually:
To manage CQL mapper and initialize the CQL schema and Elasticsearch indices, we use a Micronaut bean ElassandraStorage that use the Elasticsearch REST Client java API. To put it simply, we have run an Elassandra discover
through the Create Index API to create the Elasticsearch index baskets with a mapping automatically generated from the CQL schema.
Elassandra closely integrates the Elasticsearch code and since version 6.2.3.11+, support for Elasticsearch query over the Cassandra driver is opensource, meaning that you can query Elasticsearch through the various CQL driver implementations, with several advantages:
- Reuse the same DAO’s in your application when retrieving data from Cassandra or Elasticsearch.
- No JSON overhead, query results are sent back to the application in binary.
- The CQL paging automatically manages Elasticsearch scrolling, and the last page close the scroll context.
- The CQL driver acts as a load balancer and know about load and availability of Elassandra nodes.
- When authentication is enabled, the CQL driver manage authentication at a session level while per HTTP request authentication involve an overhead.
To send the Elasticsearch search requests to an Elassandra coordinator node, we need two dummy Cassandra columns es_query and es_options. The Elasticsearch results comes back as Cassandra rows:
The Cassandra Accessors annotation provides a nice way to map such custom queries. The CQL LIMIT clause manage the number of results returned, equivalent to the Elasticsearch query size.
A static helper method based on the Elasticsearch REST High-Level
API provides an easy way to build Elasticsearch queries, here a boolean query with two clauses, a term query and a nested query.
Micronaut supports any framework that implements Reactive Streams, including RxJava, and Reactor. As said in the documentation, if your controller method returns a non-blocking type then Micronaut will use the Event loop thread to subscribe to the result.
In our Micronaut basket controller, ListenableFutures returned by the Cassandra driver are converted to reactive types such as Single or Observable.
The TransformedListenableFuture wraps the ListenableFuture<Result<X>> to convert the result by applying a mapper function, here Result::all.
The Micronaut Testing Framework extensions included support for JUnit 5, the next generation of JUnit. To test our Cassandra and Elasticsearch queries, we wanted to use Elassandra-Unit to run an embedded Elassandra node during unit tests.
Nevertheless, in order to use the Elassandra-Unit based on Junit 4,
we need to implements some Junit 5 extensions to trigger before and after test operations.
First, an ElassandraCQLUnit5 extension to start an embedded Elassandra node where we set the Cassandra system property cassandra.custom_query_handler_class to enable support for Elasticsearch query over CQL. This could be done in the build.gradle, but it won’t be set when launching tests from IntelliJ IDEA, so this is more practical like that.
Then, our BasketControllerTest also implements a JUnit5 extension to open and cleanup Elassandra node before and after each tests. The testElassandraStorage tests our elasticsearch nested query on the baskets index.
Elassandra-Unit (like Cassandra-Unit developed by Jeremy Sevellec)
use the CQL port 9142 (not the default 9042/tcp) and cluster name ”Test Cluster” (defined in a resource file of elassandra-unit),
the test/resources/application.yaml:
Finally, tests are successful with our embedded Elassandra node:
As a take away, the gradle JIB plugin quickly containerize the basketapp
application and publish the docker image to a docker registry:
./gradlew clean jib — image strapdata/basketapp:0.1
Then, deploy the basketapp on Kubernetes with a service and a deployment:
You can also deploy Elassandra (Using the Elassandra HELM Chart), Kibana for reporting and Traefik to expose the basketapp service:
Et voilà, you get a reliable, reactive and efficient REST micro-service backed by Elassandra. Data integration tests are very useful in terms of making sure that our code runs correctly up to the database, and Elassandra-Unit helps you to check both Cassandra and Elasticsearch queries.
Next step is to run automated integration tests, and Kubernetes can help to dynamically create a whole environment and discard it afterward. You can have a look at the Elassandra HELM charts for that.
Finally, this architecture is easy to scale by adding nodes (app or Elassandra pods), always up during node failures or rolling upgrades. No more database to Elasticsearch synchronization headache, Elassandra properly index your Cassandra data into Elasticsearch !
