java - Apache Storm 中的单元测试 - 使用 BaseRichBolt 超时，但使用 BaseBasicBolt 则不超时

Question

我正在尝试为 Storm Bolts (Java) 实现单元测试。下面的代码运行良好，并在 Storm 1.0.3 上成功:

测试运行:1，失败:0，错误:0，跳过:0，已用时间:3.887 秒

但是，当我将第 46 行上的 BaseBasicParrotBolt 更改为 BaseRichParrotBolt 时，断言永远不会运行，并以以下异常结束:

13610 [main] 错误 o.a.s.testing4j - 集群 java.lang.AssertionError 中的错误:测试超时(10000ms)(不是(每个？耗尽？(spout-objects spouts)))

如果您使用调试器单步执行它，您将看到 Bolt 确实接收和发出元组，但似乎 Testing.completeTopology 永远不会返回。我觉得这很奇怪，因为 bolt 几乎是相同的。我的所有 bolt 均从 BaseRichBolt 延伸，因此我真的很想让它也适用于这些 bolt 。有什么想法吗？

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

Answer 1

如果使用 BaseRichBolt ，您应该在execute()中自己调用 ack() ，这由 BaseBasicBolt 处理。