在Storm TrackedTopologyunit testing中运行Trident拓扑
如何运行Trident拓扑的JUnit测试，以便在测试和validation每个阶段的输出时允许元组流经拓扑？ 我已经尝试在Storm的测试框架内运行，但它不能实现Trident的validation和一致执行。
这是一个示例拓扑，其中包含一些内联注释，其中我遇到的问题最多。
import static org.junit.Assert.assertEquals; import java.util.Arrays; import java.util.List; import org.junit.Test; import storm.trident.TridentState; import storm.trident.TridentTopology; import storm.trident.operation.builtin.Count; import storm.trident.testing.MemoryMapState; import storm.trident.testing.Split; import backtype.storm.Config; import backtype.storm.ILocalCluster; import backtype.storm.Testing; import backtype.storm.testing.FeederSpout; import backtype.storm.testing.TestJob; import backtype.storm.testing.TrackedTopology; import backtype.storm.tuple.Fields; import backtype.storm.utils.Utils; public class WordCountTopologyTest { @Test public void testWordCountTopology() throws Exception { Testing.withTrackedCluster(new WordCountTestJob()); } public class WordCountTestJob implements TestJob { @Override public void run(ILocalCluster cluster) throws Exception { // Create the test topology to submit TridentTopology termCountTopology = new TridentTopology(); FeederSpout feeder = new FeederSpout(new Fields("text", "author")); TridentState tridentState = termCountTopology.newStream("inputSpout", feeder) .each(new Fields("text"), new Split(), new Fields("word")) .groupBy(new Fields("word")) .name("counter-output") .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count")) .parallelismHint(6); TrackedTopology tracked = Testing.mkTrackedTopology(cluster, termCountTopology.build()); // Feed some random data into the topology feeder.feed(Arrays.asList("Nearly all men can stand adversity, but if you want to test a man's character, give him power.", "Abraham Lincoln")); feeder.feed(Arrays.asList("No man has a good enough memory to be a successful liar.", "Abraham Lincoln")); feeder.feed(Arrays.asList("Either write something worth reading or do something worth writing.", "Benjamin Franklin")); feeder.feed(Arrays.asList("Well done is better than well said.", "Benjamin Franklin")); cluster.submitTopology("word-count-testing", new Config(), tracked.getTopology()); // (!!) Runs, but bad to sleep for any time when may run faster or slower on other systems // Utils.sleep(5000); // (!!) Fails with 5000ms Topology timeout // Testing.trackedWait(tracked, 3); /* * (!!) Always 0. Trident creates the streams and bolts internally with * different names, so how can we read them to verify? */ List outputTuples = Testing.readTuples(tracked, "counter-output"); assertEquals(0, outputTuples.size()); } } } 除此之外，我已经尝试编写自己的BaseFilter来标记存储所有元组的流的末尾，但似乎必须有更好的方法。 而且，这不能解决以受控方式运行拓扑的问题。 这是Trident支持的东西吗？
使用FeederBatchSpout类（对于Trident）而不是FeederSpout。 FeederBatchSpout单独阻塞，不需要使用Testing.trackedWait（）或类似的东西。
来源： https ： //groups.google.com/forum/#!topic/storm- user/ CrAdQEXo5OU