StormKafkaSpout extends BaseRichSpout implements Configurable { @Override public void nextTuple() { if (isRateLimited()) { rateLimiter.acquire(); } kafkaSpout.nextTuple(); }  @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 testNextTupleWithRateLimiter() throws Exception { spout.rateLimiter = rateLimiter; spout.kafkaSpout = kafkaSpout; new StrictExpectations() {{ rateLimiter.acquire(); kafkaSpout.nextTuple(); }}; spout.permitsPerSecond = 1D; spout.nextTuple(); }
@Test public void testNextTupleWithoutRateLimiter() throws Exception { spout.rateLimiter = rateLimiter; spout.kafkaSpout = kafkaSpout; new StrictExpectations() {{ kafkaSpout.nextTuple(); }}; spout.permitsPerSecond = spout.DEFAULT_PERMITS_PER_SECOND; spout.nextTuple(); new Verifications() {{ rateLimiter.acquire(); times = 0; }}; }