How to use the Testing Api - xumingming/storm GitHub Wiki

Storm has a great suite of testing facilities, it is used for unit-test the Storm itself, previously its only available in Clojure, now it's also available in Java, storm users can use it unit test your own topologies. This article will introduce these apis.

Overview

These apis are encapsulated in class: backtype.storm.Testing, includes

• withSimulatedTime
• withLocalCluster
• withSimulatedTimeLocalCluster
• completeTopology
• withTrackedCluster
• readTuples
• submitLocalTopology
• mkTrackedTopology
• trackedWait
• advanceClusterTime
• eq

We will cover every api in the following sections:

withSimulatedTime

In storm, some logic relies on the time, for example the Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, its not wise and waste time if we actually to wait some time to test our topology, so we provided the time simulation facility:

Testing.withSimulatedTime(new Runnable() {
	@Override
	public void run() {
	    assertTrue(Time.isSimulating());
            // advance the time one second
            Time.advanceTime(1000);
	}
});

you invoke the withSimulatedTime method by provide a Runnable, then the code in Runnable will be running in a simulated time environment.

withLocalCluster

In unit testing, we need to utilize the LocalCluster to test our topology, but the annoying thing is to use it, you need to first create the cluster, and at the end, you need to shut it down:

LocalCluster cluster = new LocalCluster();
....
cluster.shutdown();

withLocalCluster will make it more confortable to do unit test with LocalCluster, you dont need to create, you dont need to shutdown, just use the cluster we give you!

MkClusterParam mkClusterParam = new MkClusterParam();
mkClusterParam.setSupervisors(2);
mkClusterParam.setPortsPerSupervisor(5);
Config daemonConf = new Config();
daemonConf.put(Config.SUPERVISOR_ENABLE, false);
daemonConf.put(Config.TOPOLOGY_ACKER_EXECUTORS, 0);

Testing.withLocalCluster(mkClusterParam, new TestJob() {
	@Override
	public void run(ILocalCluster cluster) {
		assertNotNull(cluster.getState());
	}
});

To use withLocalCluster, you provide a MkClusterParam which config the property of the cluster and a TestJob in which your testing code will run, the run method will give you a ILocalCluster to use.

completeTopology

When do unit-test, we always do some kind of mocking to mock the environment which is not relevant to what we are going to test, completeTopology gives you the ability for topology, the method signature is:

Map completeTopology(ILocalCluster cluster, StormTopology topology, CompleteTopologyParam param)
Map completeTopology(ILocalCluster cluster, StormTopology topology)

The main things we can config is in the CompleteTopologyParam class:

• Config stormConf -- the config for the topology
• MockedSources mockedSources -- we put the mocked tuples in this object
• boolean cleanupState -- whether to cleanup state in the end
• String topologyName -- the topologyName for your topology

e.g.