Using Hystrix with Dropwizard

I've previously blogged about Hystrix/Tenacity and Breakerbox. The full code and outline of the example is located here.

Subsequently I've been using Dropwizard and Hystrix without Tenacity/Breakerbox and found it far simpler. I don't see a great deal of value in adding Tenacity and Breakerbox as Hystrix uses Netflix's configuration library Archaius which already comes with dynamic configuration via files, databases and Zookeeper.

So lets see what is involved in integrating Hystrix and Dropwizard.

The example is the same from this article. Briefly, it is a single service that calls out to three other services:

• A user service
• A pin check service
• A device service

To allow the application to reliably handle dependency failures we are going to call out to each of the three services using Hystrix commands. Here is an example of calling out using the Apache Http Client to a pin check service:

To execute this command you simply instantiate an instance and call execute(), Hystrix handles creating a work queue and thread pool. Each command that is executed with the same group will use the same work queue and thread pool. You tell Hystrix the group by passing it to super() when extending a Hystrix command. To configure Hystrix in a Dropwizard like way we can add a map to our Dropwizard YAML:

This will translate to a Map in your Dropwizard configuration class:


The advantage of using a simple map rather than a class with the property names matching Hystrix property names is this allows you to be completely decoupled from Hystrix and its property naming conventions. It also allows users to copy property names directly from Hystrix documentation into the YAML.

To enable Hystrix to pick these properties up it requires a single line in your Dropwizard application class. This simplicity is due to the fact that Hystrix uses Archaius for property management.


Now you can add as any of Hystrix's many properties to your YAML. Then later extend the Configuration you install to include a dynamic configuration source such as ZooKeeper.

I hope this shows just how simple it is to use Hystrix with Dropwizard without bothering with Tenacity. A full working example is on github. 

Stubbed Cassandra at Skills Matter

Yesterday I gave a talk on how to test Cassandra applications using Stubbed Cassandra at the Skills Matter in London for the Cassandra London meetup group.

The talk was well attended with some where between 50 to 100 people attending.

The slides are on Slide share:


And the talk is on the skills matter website.

Thanks to Cassandra London and Skills Matter for having me!

RabbitMQ and Highly Available Queues

RabbiqMQ is a AMQP broker with an interesting set of HA abilities. Do a little research and your head will start spinning working out the differences between making messages persistent, or queues durable, or was it durable messages and HA queues with transactions? Hopefully the following is all the information you need in one place.

Before evaluating them you need to define your requirements.

• Do you want queues to survive broker failures? 
• Do you want unconsumed messages to survive a broker failure?
• What matters more, publisher speed, or the above? Or do you want a nice compromise?

RabbitMQ allows you to:

• Make a cluster of Rabbits where clients can communicate with any node in the cluster
• Make a queue durable, meaning the queue definition itself will survive broker failure
• Make a message persistent, meaning that it will get stored to disk, which you do by setting a message's delivery_mode
• Make a queue HA, meaning its contents will be replicated across brokers, either a specified list, all of them or a number of them 
• Even an HA queue has a single master that handles all operations on that queue even if the client is connected to a different node in the cluster, the master sends information to the replicas, these are called slaves

Okay so you have a durable queue that is HA and you're using persistent messages (you really want it all!). How do you work with the queue correctly?

Producing to an HA queue

You have three options for publishing to a HA queue:

• Accept the defaults, the publish will return with no guarantees in the result of broker failure
• Publisher confirms
• Transactions

The defaults: You went to all that effort of making a durable HA queue and send a persistent message and then you just fire and forget? Sounds crazy, but its not. You might have done the above to make sure you don't lose a lot of messages, but you don't want the performance impact of waiting for any form of acknowledgment. You're essentially accepting a few failures when you lose a rabbit that is the master for any of your queues.

Transactions: To use RabbitMQ transactions you do a txSelect on your channel. Then when you publish a message you call txCommit which won't return until your message has been accepted by all of the master and all of the queues slaves. If you message is persistent then that means it is on the disk of them all, you're safe! What's not to like? The speed! Every persistent message that is published in a transaction results in an fsync to disk. You need a compromise you say? 

Publisher confirms: So you don't want to lose your messages and you want to speed things up. Then you can enable publish confirms on your channel. RabbitMQ will then send you a confirmation when the message has made it to disk on all the rabbits but it won't do it right away, it will flush things to disk in batches. You can either block periodically or set up a listener to get notified. Then you can put logic in your publisher to do retries etc. You might even write logic to limit the number of published messages that haven't been confirmed. But wait, isn't queueing meant to be easy?

Consuming from a HA queue

Okay, so you have your message on the queue - how do you consume it? This is simpler:

• Auto-ack: As soon as a message is delivered RabbitMQ discards it
• Ack: Your consumer has to manually ack each message

If your consumer crashes and disconnects from Rabbit then the message will be re-queued. However if you have a bug and you just don't ack it, then Rabbit will keep a hold of it until you disconnect, then it will be re-queued. I bet that leads to some interesting bugs!

So what could go wrong?

This sounds peachy, you don't care about performance so you have a durable HA queue with persistent messages and are using transactions for producing and acks when consuming, you guaranteed exactly once delivery right? Well, no. Imagine your consumer crashes having consumed the message but just before sending the ack? Rabbit will re-send the message to another consumer.

HA queueing is hard!

Conclusion 

There is no magic bullet, you really need to understand the software you use for HA queueing. It is complicated and I didn't even cover topics like network partitions. Rabbit is a great piece of software and its automatic failover is really great but every notch you add on (transactions etc) will degrade your performance significantly.

Getting started with Hystrix and Tenacity to build fault tolerant applications

Applications are becoming increasingly distributed. Micro service architecture is the new rage. This means that each application you develop has more and more "integration points".

Any time you make a call to another service or database, or use any third party library that is a black box to you, it can be thought of as an integration point.

Netflix's architecture gives a great example of how to deal with integration points. They have a popular open-source library called Hystrix which allows you to isolate integration points by executing all calls in its own worker queue and thread pool.

Yammer have integrated Hystrix with Dropwizard, enabling enhancement of applications to publish metrics and accept configuration updates.

Here is an example application that calls out to three HTTP services and collects the results together into a single response.

Rather then calling into a HTTP library on the thread provided by Jetty this application uses Yammer's Hystrix wrapper, Tenacity.

Let's look at one of the integration points:

Here we extend the TenacityCommand class and call the dependency in the run() method. Typically all this code would be in another class with the TenacityCommand just being a wrapper, but this is a self-contained example. Let's explain what it is doing:

• Making an HTTP call using the Apache HTTP client
• If it fails throw a Runtime exception

By instantiating this TenacityCommand and calling execute(), your code is automagically executed on its very own thread pool, and requests are queued on its very own work queue. What benefits do you get?

• You get a guaranteed timeout, so no more relying on library's read timeout that never seems to work in production
• You get a circuit breaker that opens if a configured % of calls fail, meaning you can fail fast and throttle calls to failing dependencies
• You get endpoints that show the configuration and whether a circuit breaker is open

If the call to run() fails, times out or the circuit breaker is open Tenacity will call the optional getFallback() method, so you can provide a fallback strategy rather than failing completely. 

Another hidden benefit is how easy it is to move to a more asynchronous style of programming. Let's look at the resource class that pulls together the three dependencies:

Let's ignore the fact we are calling out to other HTTP services from a resource layer. The above code shows how to use Tenacity synchronously. Apart from the advantages you gain regarding failures, all the calls are still happening one by one as we call execute() which blocks so we don't call the second dependency until the first one has finished.

However, this doesn't have to be the case. Now you've snuck Tenacity into your code base you can change the code to something like this:

And without your colleagues realising you've made all of your calls to your dependencies execute asynchronously and (possibly) at the same time, then you block to bring them all together at the end.

We've barely scratched the surface of Hystrix and Tenacity but hopefully you can already see the benefits. All the code for this example a long with instructions on how to use wiremock to mock the dependencies is here.

Rabbit MQ vs Apache QPID - picking your AMQP broker

Two of the top AMQP brokers are RabbitMQ and QPID. Which one is for you? Here are some of the considerations:

Both have the same programming model as you can connect with any AMQP compatible client, so how do you pick?

First off you need to decide if you're interested in the Java QPID broker or the C++ QPID broker. They have the same name but they are quite different. This article will compare the C++ QPID broker.

Dependencies

RabbitMQ has been developed in Erlang, and thus you need to install Erlang before you can run it. Think of this as the equivalent of needing a JVM for running something like Tomcat. Erlang is in most Linux distros packaging system so this is easy.

QPID C++ broker doesn't have a storage dependency like its Java equivalent. To enable any form of HA QPID depends on rgmanager.

Persistance

RabbitMQ uses the built in Erlang database: Mnesia. A Mnesia database can be configured to either be in RAM or disk, which allows RabbitMQ to offer in memory / disk based queues very easily. It also means the developers of RabbitMQ haven't had all the complexities of developing a file based storage mechanism and don't rely on a database.

QPID stores its queues in memory or database. Without looking at the code that is all we're going to learn. If you see QPID referencing databases then you are looking at the QPID Java broker.

HA Queues and Failover

RabbitMQ runs in an active-passive mode for HA queues. Each queue has a master and writes are replicated to slaves. By default, RabbitMQ doesn't try and solve network partition within clusters. It expects a reliable LAN and should never be used across a WAN. However it has two optional ways of dealing with network partitions: pause_minority and autoheal. If you enable pause minority a rabbit in a cluster will disable its self by disconnecting clients if it can only communicate with a minority of nodes. The idea here is that clients will reconnect to the majority part of the cluster. Autoheal lets the nodes carry on as normal and when the network partition ends resets the nodes in the side that had the fewest client connections, how brutal!


QPID has overwhelming number of options when it comes to HA. Lets stick to in LAN HA. There you have active-passive messaging clusters. Here queues are replicated to a passive QPID node, then when the master fails you can either move a virtual IP or configure the client with both the master and the salve and it will failover. The big caveat is that QPID won't handle failover its self, you need a cluster manager like Rgmanager. The rational is that they want to avoid split brain, I'd rather have the functionality built in and not use it if I need to worry about split brain.

Cross DC? WAN compatible?

Who runs in a single DC these days? Not me! Most queueing solutions shy away from the WAN connection. Most have a clustering mode that will only work in a reliable LAN. When it comes to getting your data from one datacenter to another it is usually a bolt on.

RabbitMQ has two plugins to handle this. Lets take them individually. The Shovel plugin allows you to take messages from a queue in one cluster and put it on an exchange in another, perfect! It can be configured to create the queues/exchanges and handles connection failures. You can set properties like the ack mode depending on how resilient you want the putting of messages to be. You can also list multiple brokers so you can have it failover to another broker if you lose connection to the original. A great feature of shovels is that you can define them on every node of your cluster and it will run on a single node, if that node fails then it will start on another node.

RabbitMQ also has the federation plugin. This plugin allows you to receive messages from exchanges and queues from other brokers. It achieves the same goal as the shovel plugin but is slightly less configurable.

QPID has a very similar feature to shovels called broker federation where you define a source queue on one broker and a destination exchange on a broker running in a different exchange. Like RabbitMQ these "routes" can be replicated on a cluster of brokers and if the broker that is executing the route goes down another in the cluster can take its place.

It is a draw! However I would always consider a simpler option for cross WAN queueing. That is get the message in a queue in the datacenter is is produces but use a consumer from an application running in the other datacenter. After all it is your application that needs the message and is best places to handle failures.

Subjectiveness

RabbitMQ is far simpler to get going with, primarily due to fantastic documentation and very easy install process. I also like the the fact RabbitMQ uses Erlang's in built database and internode communication mechanism. The two modes for network partitions are very nice also. So RabbitMQ wins this tight race.

However, before you go and use RabbitMQ, checkout Apache Kafka as the wildcard!

Installing Cassandra and Spark with Ansible

I am currently doing a proof of concept with Spark and Cassandra. I quickly need to be able to create and start Cassandra and Spark clusters. Ansible to the rescue!

I split this my ansible playbook into three roles:

• Cassandra
• Ops center
• Spark

My main playbook is very simple:

I have some hosts defined in a separate hosts file called m25-cassandra. I've decided to install htop, I could have out this in a general server role.

I also define a few variables, these of course course could be defined else where per role:

• cluster_name - this will replace the cluser name in each of the hosts cassandra.yaml
• seeds - as above

So lets take a look at each role.

Cassandra

Here are the tasks:

This is doing the following:

• Installing a JRE
• Adding the Apache Cassandra debian repository
• Adding the keys for the debian repository
• Installing the latest version of Cassandra
• Replacing the cassandra.yaml (details later)
• Ensuring Cassandra is started

The template cassandra.yaml uses the following variables:

• cluster_name: '{{ cluster_name }}' - So we can rename the cluster
• - seeds: "{{ seeds }}" - So when we add a new node it connects to the cluster
• listen_address: {{ inventory_hostname }} - Listen on the nodes external IP so other nodes can communicate with it
• rpc_address: {{ inventory_hostname }} - So we can connect ops center and cqlsh to the nodes

Magic! Now adding new hosts to my hosts file with the tag m25_cassandra will get Cassandra installed, connected to the cluster and started.

Ops Center

The tasks file for ops center:

This is doing the following:

• Adding the Datastax community debian repository
• Adding the key for the repo
• Installing Ops Center
• Starting Ops Center

No templates here as all the default configuration is fine.

Spark

The spark maven build can build a debian package but I didn't find a public debian repo with it in so the following just downloads and unzips the Spark package:

I start the workers using the start-slaves.sh script from my local master do don't need to start anything on the nodes that have Cassandra on.

Conclusion

Ansible makes it very easy to install distributed systems like Cassandra. The thought of doing it manually fills me with pain! This is just got a PoC, I don't suggest downloading Spark from the public internet or always installing the latest version of Cassandra for your production systems. The full souce including templates and directory structure is here.

Behaviour driven testing with Cucumber-JVM and Stubbed Cassandra

So you're tasked with building a service that facades your company's legacy user management system. You need to do this as your part of the company needs to handle the legacy system going down.

To add resiliency you are required to save a user's information in case the legacy system is down. You need to be able to handle restarts so this cache will need to be persistent.

You're going to use Cassandra as your persistence, so how do test this? You sit down your your analyst and QA and come up with the following feature:

How would you implement these features? Assuming that the legacy system is a HTTP service you can use Wiremock to mock it being up and down.

For example here is how to mock the legacy system being up with Wiremock:

And an example of it being down:

So the next requirement is that Cassandra being down doesn't make your service fail i.e.

The first one you could stop Cassandra, perhaps using the great tool CCM. However this is slow, and you need to write code to make sure it is back up/down, all of this in a different process. And how about the next test? How do we make Cassandra return the result slowly? Or produce a Write Timeout Exception?

That is where Stubbed Cassandra comes in handy. To get is started you can add some code like this to start it before the tests and close it after tests:

Now implementing the step definition to mimic Cassandra being down is as easy as:

Which is a lot quicker than turning off a real Cassandra, and starting it back up in the @Before is also very quick.

Now to mimic Write time outs in Cassandra:

Very similar things could be done for read timeouts and unavailable exceptions.

This article gave you an insight to how you can behaviour drive features relating to Cassandra being down. The full code and running tests are here. Full information on how to use Stubbed Cassandra is here, you'll probably want to documentation for the Java client for Scassandra which is here.