StormKafkaSpout extends BaseRichSpout implements Configurable { @Override public void open(Map conf, final TopologyContext context, final SpoutOutputCollector collector) { setupKafkaSpout(); kafkaSpout.open(conf, context, collector); if (isRateLimited()) { rateLimiter = RateLimiter.create(permitsPerSecond); } }  @Override void configure(Configuration conf); boolean isRateLimited(); @Override void open(Map conf, final TopologyContext context, final SpoutOutputCollector collector); @Override void close(); @Override void nextTuple(); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override void activate(); @Override void deactivate(); @Override void declareOutputFields(OutputFieldsDeclarer declarer); static final String ZOOKEEPER; static final String ZK_ROOT; static final String BROKER_PATH; static final String TOPIC; static final String SPOUT_ID; static final String FORCE_FROM_START; static final String PERMITS_PER_SECOND; }
@Test public void testOpen( final @Injectable SpoutConfig spoutConfig, final @Injectable Map conf, final @Injectable TopologyContext context, final @Injectable SpoutOutputCollector collector, final @Injectable KafkaSpout kafkaSpout) throws Exception { spout.rateLimiter = null; spout.kafkaSpout = kafkaSpout; new Expectations(spout) {{ spout.setupKafkaSpout(); }}; spout.permitsPerSecond = spout.DEFAULT_PERMITS_PER_SECOND; spout.open(conf, context, collector); Assert.assertNull(spout.rateLimiter); spout.permitsPerSecond = 1D; spout.open(conf, context, collector); Assert.assertNotNull(spout.rateLimiter); }