Testing Storm Bolts and Spouts

Question

This is a general question regarding bolts and pins for testing modules in a storm topology written in Java.

What is the recommended practice and guide for Unit-testing (JUnit?) Bolts and drainage devices?

For example, I could write a JUnit test for Bolt , but without a complete understanding of the structure (for example, the Bolt life cycle) and the consequences of serialization, it is easy to make the mistake of creating constructor-based non-serializable member variables. In JUnit, this test will pass, but in the topology it will not work. I fully imagine that you need to consider a lot of test points (for example, this example with serialization and a life cycle).

Therefore, is it recommended that you use unit tests based on JUnit, do you run a small topology topology ( LocalMode ?) And check the implied contract for Bolt (or Spout ) in this topology? Or, is it okay to use JUnit, but does it mean that we have to simulate the Bolt life cycle (creating it, calling prepare() , mocking Config , etc.)? In this case, what common test points for the tested class (Bolt / Spout) should be considered?

What have other developers done to create the right unit tests?

I noticed that there is a topology testing API (see https://github.com/xumingming/storm-lib/blob/master/src/jvm/storm/TestingApiDemo.java ). Is it better to use any of these APIs and get “test topologies” for each individual Bolt and Spout (and check the implicit contract that Bolt should provide, for example, announced outputs)?

thank

Answer 1

Our approach is to use the constructor-injection of a serializable factory in the nozzle / bolt. Then the ejector / bolt will consult the factory in the open / prepare method. The factory’s sole responsibility is to encapsulate the receipt of spout / bolt dependencies in a serializable manner. This design allows our unit tests to introduce fake / test / prototype plants, which, when consulted, turn to prototype services. So we can narrowly push unit test / bolts using mocks, for example. Mockito.

Below is a general example of a bolt and a test for it. I omitted the factory implementation of UserNotificationFactory because it depends on your application. You can use service locators to get services, a serialized configuration, a configuration available for HDFS, or any way to get the right services if the factory can do this after the serde loop. You should cover serialization of this class.

Bolt

 public class NotifyUserBolt extends BaseBasicBolt { public static final String NAME = "NotifyUser"; private static final String USER_ID_FIELD_NAME = "userId"; private final UserNotifierFactory factory; transient private UserNotifier notifier; public NotifyUserBolt(UserNotifierFactory factory) { checkNotNull(factory); this.factory = factory; } @Override public void prepare(Map stormConf, TopologyContext context) { notifier = factory.createUserNotifier(); } @Override public void execute(Tuple input, BasicOutputCollector collector) { // This check ensures that the time-dependency imposed by Storm has been observed checkState(notifier != null, "Unable to execute because user notifier is unavailable. Was this bolt successfully prepared?"); long userId = input.getLongByField(PreviousBolt.USER_ID_FIELD_NAME); notifier.notifyUser(userId); collector.emit(new Values(userId)); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields(USER_ID_FIELD_NAME)); } } 

Test

 public class NotifyUserBoltTest { private NotifyUserBolt bolt; @Mock private TopologyContext topologyContext; @Mock private UserNotifier notifier; // This test implementation allows us to get the mock to the unit-under-test. private class TestFactory implements UserNotifierFactory { private final UserNotifier notifier; private TestFactory(UserNotifier notifier) { this.notifier = notifier; } @Override public UserNotifier createUserNotifier() { return notifier; } } @Before public void before() { MockitoAnnotations.initMocks(this); // The factory will return our mock `notifier` bolt = new NotifyUserBolt(new TestFactory(notifier)); // Now the bolt is holding on to our mock and is under our control! bolt.prepare(new Config(), topologyContext); } @Test public void testExecute() { long userId = 24; Tuple tuple = mock(Tuple.class); when(tuple.getLongByField(PreviousBolt.USER_ID_FIELD_NAME)).thenReturn(userId); BasicOutputCollector collector = mock(BasicOutputCollector.class); bolt.execute(tuple, collector); // Here we just verify a call on `notifier`, but we could have stubbed out behavior befor // the call to execute, too. verify(notifier).notifyUser(userId); verify(collector).emit(new Values(userId)); } } 

Answer 2

Since version 0.8.1, storm object test objects were opened through Java:

• http://storm-project.net/2012/09/06/storm081-released.html

An example of using this API is here:

• https://github.com/xumingming/storm-lib/blob/master/src/jvm/storm/TestingApiDemo.java

Answer 3

One approach that we have taken is to transfer most of the application logic from bolts and nozzles to the objects that we use for intensive lifting by creating instances and using them using minimal interfaces. Then we conduct unit testing of these objects and integration testing, although this leaves a gap.

Answer 4

It turns out it's pretty easy to mock objects like OutputDeclarer, Tuple, and OutputFieldsDeclarer. Of these, only OutputDeclarer ever sees any side effects, so the code of the OutputDeclarer mock class can respond to any tuples and anchors emitted, for example. Your test class can then use instances of these mock classes to easily customize the bolt / nozzle instance, call it, and check for expected side effects.