测试风暴螺栓和喷口

Question

这是关于用Java编写的风暴拓扑中的单元测试螺栓和喷口的一般性问题.

单元测试(JUnit？)螺栓和喷嘴的推荐做法和指南是什么？

例如,我可以为a编写一个JUnit测试Bolt,但是如果没有完全理解框架(如a的生命周期Bolt)和序列化含义,很容易犯错误的基于构造函数的非可序列化成员变量的创建.在JUnit中,此测试将通过,但在拓扑中,它将无法工作.我完全可以想象有许多测试点需要考虑(例如序列化和生命周期的这个例子).

因此,是否建议您使用基于JUnit的单元测试,运行小型模拟拓扑(LocalMode？)并测试该拓扑下的Bolt(或Spout)隐含合约？或者,使用JUnit是否可以,但其含义是我们必须仔细模拟Bolt的生命周期(创建它,调用prepare(),模拟Config等)？在这种情况下,要测试的被测类(Bolt/Spout)的一般测试点是什么？

在创建适当的单元测试方面,其他开发人员做了什么？

我注意到有一个拓扑测试API(参见:https://github.com/xumingming/storm-lib/blob/master/src/jvm/storm/TestingApiDemo.java).是否更好地使用某些API,并为每个人站起来"测试拓扑" Bolt&Spout(并验证Bolt必须提供的隐式合同,例如 - 它的声明输出)？

谢谢

Answer 1

从版本0.8.1开始,Storm的单元测试工具已通过Java公开:

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

有关如何使用此API的示例,请查看此处:

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

Answer 2

我们采取的一种方法是将大部分应用程序逻辑从螺栓和喷口中移出,并通过实例化和通过最小接口使用它们来进行繁重的操作.然后我们对这些对象和集成测试进行单元测试,尽管这确实留下了空白.

Answer 3

我们的方法是使用可序列化工厂的构造函数注入到喷口/螺栓中.然后,喷嘴/螺栓以其开放/准备方法咨询工厂.工厂的唯一责任是以可序列化的方式封装获取喷口/螺栓的依赖关系.这种设计允许我们的单元测试注入假/测试/模拟工厂,当被咨询时,返回模拟服务.通过这种方式,我们可以使用模具(例如Mockito)对喷嘴/螺栓进行狭窄的单元测试.

下面是一个螺栓的通用示例和它的测试.我省略了工厂的实现,UserNotificationFactory因为它取决于你的应用程序.您可以使用服务定位器来获取服务,序列化配置,HDFS可访问配置,或者实际上以任何方式获得正确的服务,只要工厂可以在serde循环后执行.您应该涵盖该类的序列化.

螺栓

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));
  }
}

测试

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));
  }
}