Apache風暴中的單元測試 - 使用BaseRichBolt超時但不使用BaseBasicBolt

Question

我正在嘗試爲風暴螺栓（Java）實現單元測試。下面的代碼工作正常，並在風暴1.0.3與成功結束：

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

Answer 1

如果使用BaseRichBolt，你應該叫ACK（）在自己的execute（），這是由BaseBasicBolt處理。