How to simulate Coherence node failure in JUnit test

Recently, I was working on renewed version of Coherence data loss listener. New version provides simple utility to attach partition loader to any cache. Such partition loader is guaranteed to be called for every partition of newly created cache or after partition has been lost. Unlike CacheStore, partition loader will be called on the node there it has been added. This way you could have dedicated set loader processes, which are not involved at storing a data. Also it is guaranties that

only one instance of partition loader are executed for particular partition in cluster,

if there is at least one live node with registered partition loader it would be invoked for empty partition.

But let’s get back to a topic of post. I’m actively using JUnit and my test-utils library to automate testing of my Coherence stuff. Test-utils is using class loader isolation to run multiple Coherence nodes in single JVM.

But, unlike many other tests, here I need to test disaster case. Coherence should think that one of its nodes has died. Normally, I’m using CacheFactory.shutdown() to kill virtualized Coherence node, but this way it would be a graceful shutdown.

For data loss listener, I really want to test disaster case.

How Coherence track node liveness?

Naïve approach using timeout is not working well with data grid prioritizing resilience and performance such as Coherence.

What is problem with timeout?

If you let it be too short, there will be too many false positives making grid unstable (JVM may do a GC, OS may start swapping, etc).

If you let it be too long, time of recovery from disaster would be too long.

How this can be improved? Let’s see that kind of disaster could possibly happen with your cluster:

JVM process could be killed, crushed or just exited without shutdown.

Sever could crush or become unreachable via network.

Death of process could be easily tracked if you keep open TCP connection open. OS will close all TCP connections for dead process, so you could make very good assumption that remote process is dead.

Coherence is using so called TCP ring for that purpose. Each cluster node keeps two open TCP connections to other cluster nodes (forming a ring). If cluster detects that both TCP connections have been closed, it has very good reason to disconnect node right now and start recovery procedure.

In case of server/network failure, TCP connection will not be closed immediately. In addition to TCP ring, Coherence is using IP monitor to track reachability of IP addresses. If IP address cannot be reached by rest of nodes, cluster will not hesitate to disconnect all nodes from that IP.

This two tricks allow Coherence to detect real failure very fast, yet to be very tolerant to long GC pauses and other non fatal slowdowns.

Steps to kill node in JUnit test

In JUnit test all nodes in cluster are sharing same JVM. I cannot really kill a process. To simulate node death, I’m calling Thread.suspend() on all threads related to victim node (a feature of test-utils). This is making node totally unresponsive.

Two mechanisms above should be turned off in Coherence operational configuration. Disconnect timeout also should be set to smaller value (otherwise each test will take too long).

That is it, now I can test disaster cases for Coherence using JUnit.

Below is snippet of actual test:

@Test public void verify_parallel_init_crash_case() throws InterruptedException { final int partitions = 2000; final int timeout = 15000; CacheTemplate.usePartitionedServicePartitionCount(cluster, partitions); CohHelper.setTCMPTimeout(server(0), timeout); CohHelper.disableTcpRing(server(0)); CohHelper.setTCMPTimeout(client(0), timeout); CohHelper.disableTcpRing(client(0)); ... server(0).getCache("a-cache1"); statics().initPartitionCounter(partitions); // init Coherence nodes client(0).getCache("a-cache1"); client(1).getCache("a-cache1"); ... // attaching test partition loader attachTouchMonitor(0, 20, "a-cache1"); attachTouchMonitor(1, 20, "a-cache1"); ... Thread.sleep(500); System.out.println("Simulating crash for 2,3,4,5"); // simulating client crash, verify lock revocation client(2).suspend(); client(3).suspend(); client(4).suspend(); client(5).suspend(); // waiting for all test listeners to finish statics().waitAllLatches(); System.out.println("Latches are open"); Thread.sleep(200); // checking cache state assertAllCanaries(0, "a-cache1"); }

Here is a link to full java file in SVN.

Remote code execution in Java made trivial

SSH offers a very convenient way to execute shell scripts remotely. Code like

ssh myserver.acme.com << EOC
cd $APP_HOME
./start_my_stuff.sh
EOC

are fairly easy to write and read.

But while remote execution itself is easy, writing actual distributed code is total mess.

I’m a Java guy. I wish, I could run Java code remotely as easy as I can do it with shell.
And now, finally, I can.

@Test public void remote_hello_world() throws InterruptedException { ViManager cloud = CloudFactory.createSimpleSshCloud(); cloud.node("myserver.acme.com"); cloud.node("**").exec(new Callable<Void>() { @Override public Void call() throws Exception { String localHost = InetAddress.getLocalHost().toString(); System.out.println("Hi! I'm running on " + localHost); return null; } }); // Console output is transfered asynchronously, // so it is better to wait few seconds for it to arrive Thread.sleep(1000); }

And that snippet will work for you too, only two requirements are:

You should be able to SSH to target host without password (you have SSH key-pairs set up).

Remote host should have JDK installed, and java command on PATH.

That is it, there is no need to install anything special to target servers.

You can find more details in tutorial at GridKit site.

How it work?

A lot of black magic is happening behind the scene. In particular:

Classpath of current Java processes is replicated and cached at remote host using SFTP.

SSH is used to start remote Java process.

All communications between master and slave are tunneled in stdIn and stdOut (thus you can care less about NAT and firewall).

Modified version of RMI/serialization is used to allow anonymous classes to be run remotely.

By despite all that internal complexity, it just works. I’m using it to run agents on Linux and Solaris. I was using that stuff on Amazon EC2 and I cloud have master process running on my Windows desktop while slave scattered across Unixes.
It was long road. A lot of issues such as firewalls, bugs in JSch (java SSH client), subtle SSH limitation, etc has been solved along that path.
But now, I believe, it will “just work” for you too.

Java vs shell

But what is the need for such library in a first place? Below are just few of my reasons:

I’m a Java guy, I do not want to invest much in shell scripting skills.

Java is "really" cross platform, heck I can even debug stuff on Windows then run them on Linux.

I have to do a lot to distributed orchestration (starting/stopping processes in right order etc), it is so much easier to do in java.

It is hard to write reusable shell scripts, but easy for java (heck, I can even unit test pieces of deployment logic).

Remotting is just a remotting

Ability to effortlessly run remote java code is not much by itself. But it enables you to reach new levels of day-to-day automation (which was prohibitively expensive before).

But that would be topic for another post.

Coherence 101, Beware of cache listeners

Cache events facility is a quite useful feature of Oracle Coherence. For example, continuous queries and near cache features are build on top of cache event system.

Unfortunately it could be also abused easily. In particular, they are noticeably bad at scale unless you are very careful.

Please note. This article is covering only partitioned cache topology (distributed cache scheme).

Client side map listeners

UPDATE: I was very wrong in my previous description of client side synchronous map listeners. Section below was rewritten to reflect more accurate picture.

Client side map listeners are usually added via NamedCache API. They typically receive events from caches hosted on remote JVMs (storage nodes). But regardless of whenever cache event is produced at remote or local JVM, Coherence will deliver it to listeners using dedicated event dispatch thread (or service thread itself for listeners marked as synchronous).

Each cache service has only one event dispatch thread, and it could easily become a bottle neck, limiting speed of cache event processing on client.

Few tips to mitigate this design aspect are below.

• Do not do anything time consuming in listener itself, offload processing to other thread instead.
• Be careful with synchronization – avoid lock contention in listener code.
• When event hits your listener, its data are still in binary form. To avoid deserialization cost, do not access key or value in event dispatch thread, instead pass reference to map event object to own processing thread (or thread pool).

Last advice may not be intuitive, but deserialization of map event in Coherence’s event dispatch thread often becomes a bottleneck slowing down event processing rate.

Synchronous and normal map listeners

There is a marker interface SynchronousListener in com.tangosol.util package.
You could implement it in your map listener. But this wouldn’t make map event delivery to your listener synchronous with cache operation (as you may think), instead it would affect in which thread your listener is invoked.

Normal listeners are invoked in event dispatch thread.

“Synchronous” listeners would be invoked in service thread

What are the differences?

• Imagine you have near cache and you are using entry processor to update entry. If cache event would be processed in event dispatch thread, data in near cache may remain stale for short time between entry processor call have returned, but event is not processed yet.
  Using of synchronous listeners would solve this, because event would be guaranteed to be processed before processing response message from entry processor invocation.
• Time consuming custom map listeners could slow down event dispatch thread increasing event delays. This would affect Coherence build-in facilities such as near caches and CQC would be affected because they use synchronous listeners internally - you can consider it extra level of protection from misbehaving developer :)

But let me stress it again, for any type of listener events are delivered asynchronously relative to other cluster nodes.

Backing map listeners

Backing map listeners are used less often (but being abused more frequently). Backing map listeners are usually configured via XML cache configuration and work on storage side.

On storage side, Coherence could use pool of worker threads to perform operations in parallel. You may assume that you backing map listener would also be invoked in parallel …
… but that is wrong. Backing map listener could process one map event at time for given cache, regardless of thread pool size.

First time, I was also surprised by such behavior. This is not fundamental limitation of Coherence, but all out-of-box variations of backing map use cache global lock to dispatch map event. Even for partitioned backing map Coherence will use ObservableSplittingBackingMap wrapper which is, again, using global lock.

So, if you are using backing map listeners, be aware of that limitation. Live object pattern also relay on backing mapping listener and thus limited by this scalability constraint.

Map triggers

Fortunately map triggers work as a part of cache update transaction on cache service level. In other words map trigger would not harm performance more than entry processors do.

One possible workaround for baking map listeners concurrency issue could be invocation of map listener from map trigger.