StormKafkaSpout extends BaseRichSpout implements Configurable { public boolean isRateLimited() { return Double.isFinite(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 testIsRateLimited() throws Exception { spout.permitsPerSecond = spout.DEFAULT_PERMITS_PER_SECOND; Assert.assertThat(spout.isRateLimited(), Is.is(false)); spout.permitsPerSecond = 1D; Assert.assertThat(spout.isRateLimited(), Is.is(true)); }