Apache风暴中的单元测试 - 使用BaseRichBolt超时但不使用BaseBasicBolt
问题描述:
我正在尝试为风暴螺栓(Java)实现单元测试。下面的代码工作正常,并在风暴1.0.3与成功结束:Apache风暴中的单元测试 - 使用BaseRichBolt超时但不使用BaseBasicBolt
Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 3.887 sec
然而,当我上线46改变BaseBasicParrotBolt
到BaseRichParrotBolt
,因此断言不会运行,并将其与下面的异常结束:
13610 [main] ERROR o.a.s.testing4j - Error in cluster java.lang.AssertionError: Test timed out (10000ms) (not (every? exhausted? (spout-objects spouts)))
如果你通过它一步一个调试器,你会看到,螺栓不接收和发射元组,但它似乎是Testing.completeTopology
永远不会返回。我觉得这很奇怪,因为螺栓几乎是相同的。我所有的螺栓从BaseRichBolt
延伸,所以我真的很想让它为那些工作。有任何想法吗?
import java.util.Map;
import org.apache.storm.Config;
import org.apache.storm.ILocalCluster;
import org.apache.storm.Testing;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.testing.CompleteTopologyParam;
import org.apache.storm.testing.MkClusterParam;
import org.apache.storm.testing.MockedSources;
import org.apache.storm.testing.TestJob;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import java.util.Arrays;
import java.util.List;
import static junit.framework.Assert.*;
import org.junit.Test;
public class StormTestExample {
private final static String EVENT = "event";
private final static String SPOUT_ID = "spout";
private final static String BOLT_ID = "parrot";
private final static List<String> COMPONENT_IDS = Arrays.asList(SPOUT_ID, BOLT_ID);
@Test
public void testBasicTopology() {
MkClusterParam mkClusterParam = new MkClusterParam();
mkClusterParam.setSupervisors(4);
Config daemonConf = new Config();
daemonConf.put(Config.STORM_LOCAL_MODE_ZMQ, false);
mkClusterParam.setDaemonConf(daemonConf);
Testing.withSimulatedTimeLocalCluster(mkClusterParam, new TestJob() {
@Override
public void run(ILocalCluster cluster) {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, new TestSpout());
builder.setBolt(BOLT_ID, new BaseBasicParrotBolt()).shuffleGrouping(SPOUT_ID);
StormTopology topology = builder.createTopology();
MockedSources mockedSources = new MockedSources();
mockedSources.addMockData(SPOUT_ID,
new Values("nathan"),
new Values("bob"),
new Values("joey"),
new Values("nathan"));
Config conf = new Config();
conf.setNumWorkers(2);
CompleteTopologyParam completeTopologyParam = new CompleteTopologyParam();
completeTopologyParam.setMockedSources(mockedSources);
completeTopologyParam.setStormConf(conf);
final Map result = Testing.completeTopology(cluster, topology, completeTopologyParam);
final Values expected = new Values(new Values("nathan"), new Values("bob"), new Values("joey"),
new Values("nathan"));
for (String component : COMPONENT_IDS) {
assertTrue("Error in " + component + " output",
Testing.multiseteq(expected, Testing.readTuples(result, component)));
}
}
});
}
private static class TestSpout extends BaseRichSpout {
@Override
public void declareOutputFields(OutputFieldsDeclarer ofd) {
ofd.declare(new Fields(EVENT));
}
@Override
public void open(Map map, TopologyContext tc, SpoutOutputCollector soc) {
throw new UnsupportedOperationException(); // Don't need an implementation to run the test.
}
@Override
public void nextTuple() {
throw new UnsupportedOperationException(); // Don't need an implementation to run the test.
}
}
private static class BaseBasicParrotBolt extends BaseBasicBolt {
@Override
public void declareOutputFields(OutputFieldsDeclarer ofd) {
ofd.declare(new Fields(EVENT));
}
@Override
public void execute(Tuple tuple, BasicOutputCollector boc) {
boc.emit(new Values(tuple.getValue(0)));
}
}
private static class BaseRichParrotBolt extends BaseRichBolt {
private OutputCollector oc;
@Override
public void declareOutputFields(OutputFieldsDeclarer ofd) {
ofd.declare(new Fields(EVENT));
}
@Override
public void prepare(Map map, TopologyContext tc, OutputCollector oc) {
this.oc = oc;
}
@Override
public void execute(Tuple tuple) {
oc.emit(new Values(tuple.getValue(0)));
}
}
}
答
如果使用BaseRichBolt,你应该叫ACK()在自己的execute(),这是由BaseBasicBolt处理。
适用于Storm 1.0.3。我没有意识到这可能会对测试产生影响,很好的结果。我从来没有做过任何事情,因为对于我的用例来说,最多一次就足够了。 不幸的是,这并没有解决它的风暴0.9.6(我仍然坚持atm ...但将很快切换...) – jvlier