Merge pull request #911 from mattrjacobs/queue-saturation

Unit tests around large threadpool/small queue scenario

Author: mattrjacobs

hystrix-core/src/test/java/com/netflix/hystrix/HystrixCommandTest.java

@@ -23,6 +23,7 @@
 import static org.junit.Assert.fail;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.ArrayBlockingQueue;
@@ -4306,6 +4307,75 @@ protected Integer run() throws Exception {
         assertTrue(semaphoreExceptionEncountered.get());
     }
 
+    @Test
+    public void testCommandConcurrencyViaObserveExceedsQueueSizeButNotThreadPoolSize() {
+        List<Observable<Boolean>> cmdResults = new ArrayList<Observable<Boolean>>();
+
+        HystrixThreadPool threadPool = null;
+
+        //thread pool size is 20, so we have room for concurrent execution of all 20 commands
+        //but queue size is 2 - do we see any queue rejections? - we should not
+        for (int i = 0; i < 20; i++) {
+            HystrixCommand<Boolean> cmd = new CommandWithLargeThreadPoolSmallQueue();
+            if (threadPool == null) {
+                threadPool = cmd.threadPool;
+            }
+            cmdResults.add(cmd.toObservable());
+        }
+
+        Observable<Boolean> allObservables = Observable.merge(cmdResults);
+
+        TestSubscriber<Boolean> subscriber = new TestSubscriber<Boolean>();
+
+        allObservables.subscribe(subscriber);
+
+        subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS);
+        if (subscriber.getOnErrorEvents().size() > 0) {
+            subscriber.getOnErrorEvents().get(0).printStackTrace();
+        }
+
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(20);
+    }
+
+    @Test
+    public void stressTestLargeThreadPoolSmallQueueUsingObserve() {
+        for (int n = 0; n < 20; n++) {
+            testCommandConcurrencyViaObserveExceedsQueueSizeButNotThreadPoolSize();
+            Hystrix.reset();
+        }
+    }
+
+    @Test
+    public void testCommandConcurrencyViaQueueExceedsQueueSizeButNotThreadPoolSize() throws Exception {
+        List<Future<Boolean>> cmdResults = new ArrayList<Future<Boolean>>();
+
+        HystrixThreadPool threadPool = null;
+
+        //thread pool size is 20, so we have room for concurrent execution of all 20 commands
+        //but queue size is 2 - do we see any queue rejections? - we should not
+        for (int i = 0; i < 20; i++) {
+            HystrixCommand<Boolean> cmd = new CommandWithLargeThreadPoolSmallQueue();
+            if (threadPool == null) {
+                threadPool = cmd.threadPool;
+            }
+            cmdResults.add(cmd.queue());
+        }
+
+        for (Future<Boolean> f: cmdResults) {
+            f.get(1, TimeUnit.SECONDS);
+        }
+    }
+
+    @Test
+    public void stressTestLargeThreadPoolSmallQueueUsingQueue() throws Exception {
+        for (int n = 0; n < 20; n++) {
+            testCommandConcurrencyViaQueueExceedsQueueSizeButNotThreadPoolSize();
+            Hystrix.reset();
+        }
+    }
+
     /* ******************************************************************************** */
     /* ******************************************************************************** */
     /* private HystrixCommand class implementations for unit testing */
@@ -5176,4 +5246,17 @@ protected Boolean getFallback() {
             return false;
         }
     }
+
+    private static class CommandWithLargeThreadPoolSmallQueue extends TestHystrixCommand<Boolean> {
+
+        public CommandWithLargeThreadPoolSmallQueue() {
+            super(testPropsBuilder().setThreadPoolPropertiesDefaults(
+                    HystrixThreadPoolProperties.Setter().withMaxQueueSize(2).withQueueSizeRejectionThreshold(2).withCoreSize(20)));
+        }
+
+        @Override
+        protected Boolean run() throws Exception {
+            return true;
+        }
+    }
 }