Allow custom overrides with global filter concurrency limits

zuul-core/src/main/java/com/netflix/zuul/filters/BaseFilter.java

@@ -124,14 +124,19 @@ public HttpContent processContentChunk(ZuulMessage zuulMessage, HttpContent chun
 
     @Override
     public void incrementConcurrency() throws ZuulFilterConcurrencyExceededException {
-        final int limit = Math.max(filterConcurrencyCustom.get(), filterConcurrencyDefault.get());
+        final int limit = calculateConcurency();
         if ((concurrencyProtectionEnabled.get()) && (concurrentCount.get() >= limit)) {
             concurrencyRejections.increment();
             throw new ZuulFilterConcurrencyExceededException(this, limit);
         }
         concurrentCount.incrementAndGet();
     }
 
+    protected int calculateConcurency() {
+        final int customLimit = filterConcurrencyCustom.get();
+        return customLimit != DEFAULT_FILTER_CONCURRENCY_LIMIT ? customLimit : filterConcurrencyDefault.get();
+    }
+
     @Override
     public void decrementConcurrency() {
         concurrentCount.decrementAndGet();

zuul-core/src/test/java/com/netflix/zuul/filters/BaseFilterTest.java

@@ -24,26 +24,27 @@
 import com.netflix.zuul.message.ZuulMessage;
 import com.netflix.zuul.message.ZuulMessageImpl;
 import org.apache.commons.configuration.AbstractConfiguration;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import rx.Observable;
 
 /**
- * Tests for {@link BaseFilter}.   Currently named BaseFilter2Test as there is an existing class named BaseFilterTest.
+ * Tests for {@link BaseFilter}
  */
 @RunWith(MockitoJUnitRunner.class)
 class BaseFilterTest {
 
-    @Mock
-    private BaseFilter f1;
-
-    @Mock
-    private BaseFilter f2;
-
     @Mock
     private ZuulMessage req;
+    private final AbstractConfiguration config = ConfigurationManager.getConfigInstance();
+
+    @BeforeEach
+    public void tearDown() {
+        config.clear();
+    }
 
     @Test
     void testShouldFilter() {
@@ -84,7 +85,7 @@ class ConcInboundFilter extends BaseFilter {
 
             @Override
             public Observable applyAsync(ZuulMessage input) {
-                limit[0] = Math.max(filterConcurrencyCustom.get(), filterConcurrencyDefault.get());
+                limit[0] = calculateConcurency();
                 return Observable.just("Done");
             }
 
@@ -103,17 +104,16 @@ public boolean shouldFilter(ZuulMessage msg) {
     }
 
     @Test
-    void validateFilterConcurrencyLimitOverride() {
-        AbstractConfiguration configuration = ConfigurationManager.getConfigInstance();
-        configuration.setProperty("zuul.filter.concurrency.limit.default", 7000);
-        configuration.setProperty("zuul.ConcInboundFilter.in.concurrency.limit", 4000);
+    void validateFilterGlobalConcurrencyLimitOverride() {
+        config.setProperty("zuul.filter.concurrency.limit.default", 7000);
+        config.setProperty("zuul.ConcInboundFilter.in.concurrency.limit", 4000);
         final int[] limit = {0};
 
         class ConcInboundFilter extends BaseFilter {
 
             @Override
             public Observable applyAsync(ZuulMessage input) {
-                limit[0] = Math.max(filterConcurrencyCustom.get(), filterConcurrencyDefault.get());
+                limit[0] = calculateConcurency();
                 return Observable.just("Done");
             }
 
@@ -130,4 +130,32 @@ public boolean shouldFilter(ZuulMessage msg) {
         new ConcInboundFilter().applyAsync(new ZuulMessageImpl(new SessionContext(), new Headers()));
         Truth.assertThat(limit[0]).isEqualTo(7000);
     }
+
+    @Test
+    void validateFilterSpecificConcurrencyLimitOverride() {
+        config.setProperty("zuul.filter.concurrency.limit.default", 7000);
+        config.setProperty("zuul.ConcInboundFilter.in.concurrency.limit", 4300);
+        final int[] limit = {0};
+
+        class ConcInboundFilter extends BaseFilter {
+
+            @Override
+            public Observable applyAsync(ZuulMessage input) {
+                limit[0] = calculateConcurency();
+                return Observable.just("Done");
+            }
+
+            @Override
+            public FilterType filterType() {
+                return FilterType.INBOUND;
+            }
+
+            @Override
+            public boolean shouldFilter(ZuulMessage msg) {
+                return true;
+            }
+        }
+        new ConcInboundFilter().applyAsync(new ZuulMessageImpl(new SessionContext(), new Headers()));
+        Truth.assertThat(limit[0]).isEqualTo(4300);
+    }
 }