ServiceRequestLimiter extends DynamicRequestLimiter<StreamOp> { @Override public RequestLimiter<StreamOp> build() { int rpsStreamAcquireLimit = dynConf.getRpsStreamAcquireServiceLimit(); int rpsSoftServiceLimit = dynConf.getRpsSoftServiceLimit(); int rpsHardServiceLimit = dynConf.getRpsHardServiceLimit(); int bpsStreamAcquireLimit = dynConf.getBpsStreamAcquireServiceLimit(); int bpsSoftServiceLimit = dynConf.getBpsSoftServiceLimit(); int bpsHardServiceLimit = dynConf.getBpsHardServiceLimit(); RequestLimiterBuilder rpsHardLimiterBuilder = RequestLimiterBuilder.newRpsLimiterBuilder() .statsLogger(limiterStatLogger.scope("rps_hard_limit")) .limit(rpsHardServiceLimit) .overlimit(new OverlimitFunction<StreamOp>() { @Override public void apply(StreamOp request) throws OverCapacityException { throw new OverCapacityException("RPS limit exceeded for the service instance"); } }); RequestLimiterBuilder rpsSoftLimiterBuilder = RequestLimiterBuilder.newRpsLimiterBuilder() .statsLogger(limiterStatLogger.scope("rps_soft_limit")) .limit(rpsSoftServiceLimit); RequestLimiterBuilder bpsHardLimiterBuilder = RequestLimiterBuilder.newBpsLimiterBuilder() .statsLogger(limiterStatLogger.scope("bps_hard_limit")) .limit(bpsHardServiceLimit) .overlimit(new OverlimitFunction<StreamOp>() { @Override public void apply(StreamOp request) throws OverCapacityException { throw new OverCapacityException("BPS limit exceeded for the service instance"); } }); RequestLimiterBuilder bpsSoftLimiterBuilder = RequestLimiterBuilder.newBpsLimiterBuilder() .statsLogger(limiterStatLogger.scope("bps_soft_limit")) .limit(bpsSoftServiceLimit); ChainedRequestLimiter.Builder<StreamOp> builder = new ChainedRequestLimiter.Builder<StreamOp>(); builder.addLimiter(new StreamAcquireLimiter(streamManager, serviceRps, rpsStreamAcquireLimit, limiterStatLogger.scope("rps_acquire"))); builder.addLimiter(new StreamAcquireLimiter(streamManager, serviceBps, bpsStreamAcquireLimit, limiterStatLogger.scope("bps_acquire"))); builder.addLimiter(bpsHardLimiterBuilder.build()); builder.addLimiter(bpsSoftLimiterBuilder.build()); builder.addLimiter(rpsHardLimiterBuilder.build()); builder.addLimiter(rpsSoftLimiterBuilder.build()); builder.statsLogger(limiterStatLogger); return builder.build(); } ServiceRequestLimiter(DynamicDistributedLogConfiguration dynConf,
                                 StatsLogger statsLogger,
                                 MovingAverageRate serviceRps,
                                 MovingAverageRate serviceBps,
                                 StreamManager streamManager,
                                 Feature disabledFeature); @Override RequestLimiter<StreamOp> build();  }
@Test(timeout = 60000) public void testChainedServiceRequestLimiter() throws Exception { MockSoftRequestLimiter softLimiter = new MockSoftRequestLimiter(new CounterLimiter(1)); MockHardRequestLimiter hardLimiter = new MockHardRequestLimiter(new CounterLimiter(3)); RequestLimiter<MockRequest> limiter = new ChainedRequestLimiter.Builder<MockRequest>() .addLimiter(softLimiter) .addLimiter(hardLimiter) .build(); assertEquals(0, softLimiter.getLimitHitCount()); assertEquals(0, hardLimiter.getLimitHitCount()); limiter.apply(new MockRequest()); assertEquals(0, softLimiter.getLimitHitCount()); assertEquals(0, hardLimiter.getLimitHitCount()); limiter.apply(new MockRequest()); assertEquals(1, softLimiter.getLimitHitCount()); assertEquals(0, hardLimiter.getLimitHitCount()); limiter.apply(new MockRequest()); assertEquals(2, softLimiter.getLimitHitCount()); assertEquals(0, hardLimiter.getLimitHitCount()); try { limiter.apply(new MockRequest()); } catch (OverCapacityException ex) { } assertEquals(3, softLimiter.getLimitHitCount()); assertEquals(1, hardLimiter.getLimitHitCount()); }