48

This is a general question regarding Unit Testing Bolts and Spouts in a Storm Topology written in Java.

What is the recommended practice and guideline for unit-testing (JUnit?) Bolts and Spouts?

For instance, I could write a JUnit test for a Bolt, but without fully understanding the framework (like the lifecycle of a Bolt) and the Serialization implications, easily make the mistake of Constructor-based creation of non-serializable member variables. In JUnit, this test would pass, but in a topology, it wouldn't work. I fully imagine there are many test points one needs to consider (such as this example with Serialization & lifecycle).

Therefore, is it recommended that if you use JUnit based unit tests, you run a small mock topology (LocalMode?) and test the implied contract for the Bolt (or Spout) under that Topology? Or, is it OK to use JUnit, but the implication being that we have to simulate the lifecycle of a Bolt (creating it, calling prepare(), mocking a Config, etc) carefully? In this case, what are some general test points for the class under test (Bolt/Spout) to consider?

What have other developers done, with respect to creating proper unit tests?

I noticed 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 some of that API, and stand up "Test Topologies" for each individual Bolt & Spout (and verifying the implicit contract that the Bolt has to provide for, eg - it's Declared outputs)?

Thanks

2
  • Did you ever decide on an approach? Jun 4, 2013 at 18:43
  • Well, I read the answers below. It seems there there are some general guidelines, but nothing set in stone. I'm going to leave the question open for a tad bit longer to see if anyone else has any thoughts then close it. I like both approaches of using the Testing API (TestingApiDemo.java) as well as your answer for mocking the dependencies, @ChrisGerken.
    – Jack
    Jun 5, 2013 at 18:29

4 Answers 4

14

Since version 0.8.1 Storm's unit testing facilities have been exposed via Java:

For an example how to use this API have a look here:

3
  • 6
    Indeed a good API, although a little bit more documentation would help a lot in understanding it.
    – Barry NL
    Nov 25, 2014 at 13:09
  • Here is a blog post which might be useful: pixelmachine.org/2011/12/17/Testing-Storm-Topologies.html It explains the clojure API for testing storm, but that is basically the same as the Java API.
    – asmaier
    Dec 4, 2015 at 14:49
  • That's not a unit test, however. That's an integration test since it spins up an in-memory cluster to execute the bolt. A unit test is one which tests the smallest possible "unit", which would be the bolt's individual methods.
    – user167019
    Apr 22, 2019 at 18:59
11

Our approach is to use constructor-injection of a serializable factory into the spout/bolt. The spout/bolt then consults the factory in its open/prepare method. The factory's single responsibility is to encapsulate obtaining the spout/bolt's dependencies in a serializable fashion. This design allows our unit tests to inject fake/test/mock factories which, when consulted, return mock services. In this way we can narrowly unit test the spout/bolts using mocks e.g. Mockito.

Below is a generic example of a bolt and a test for it. I have omitted the implementation of the factory UserNotificationFactory because it depends on your application. You might use service locators to obtain the services, serialized configuration, HDFS-accessible configuration, or really any way at all to get the correct services, so long as the factory can do it after a serde cycle. You should cover serialization of that 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));
  }
}
3
  • I know this is an old post so forgive me if you've moved on from this project :) I'm wondering whether the test for verify(notifier).notifyUser(userId); passed. I am finding that the serialization and deserialization that Storm performs on the factory causes a new mock notifier to be instantiated. Thus the original mock notifier does not receive any interactions. Was this the case for you?
    – ilana917
    May 11, 2017 at 8:04
  • @ilana917 Storm and mocks should not be interacting. The purpose of this pattern is to write the code in a way that lets you test the code separately from the Storm runtime. In the test, no serialization should occur. Between the new NotifyUserBolt and bolt.prepare lines, there is no serialization in the test. In the Storm runtime, Storm would serialize the Bolt.
    – Carl G
    May 11, 2017 at 21:04
  • Thanks @carl-g, that's the direction I'm going for. I'm in the midst of a large refactor and originally was planning on running storm end to end as a kind of integration test using mocks for external connections, but unit testing our logic makes more sense.
    – ilana917
    May 14, 2017 at 8:13
11

One approach we have taken is to move most of the application logic out of bolts and spouts and into objects that we use to do the heavy lifting by instantiating and using them via minimal interfaces. Then we do unit testing on those objects and integration testing, although this does leave a gap.

3
  • 2
    while what you say is true, and a good idea, it does not get at the OP's interest in finding things like serialization problems before doing a production deployment. Aug 5, 2014 at 20:39
  • Thank you for your response! I am curious as to what you meant by "leave a gap"? Mar 9, 2015 at 21:26
  • the gap is at the interface between storm and our objects. that part isn't tested thoroughly because it only pops up in the integration tests, which is expensive to make exhaustive, so there is some connecting code that is not well covered Mar 10, 2015 at 0:38
1

It turns out to be fairly easy to mock storm objects like OutputDeclarer, Tuple and OutputFieldsDeclarer. Of those, only OutputDeclarer ever sees any side effects so code the OutputDeclarer mock class to be able to answer any tuples and anchors emitted, for example. Your test class can then use instances of those mock classes to easily configure a bolt/spout instance, invoke it and validate the expected side effects.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.