ExecutorPipeline { public void join() throws InterruptedException { exec.shutdown(); exec.awaitTermination(10, TimeUnit.MINUTES); if (next != null) next.join(); } ExecutorPipeline(int threads, int queueSize, boolean preserveOrder, Function<In, Out> processor,
                            Consumer<Throwable> exceptionHandler); ExecutorPipeline<Out, Void> add(int threads, int queueSize, final Consumer<Out> consumer); ExecutorPipeline<Out, NextOut> add(int threads, int queueSize, boolean preserveOrder,
                                                        Function<Out, NextOut> processor); void push(final In in); void pushAll(final List<In> list); ExecutorPipeline<In, Out> setThreadPoolName(String threadPoolName); BlockingQueue<Runnable> getQueue(); Map<Long, Out> getOrderMap(); void shutdown(); boolean isShutdown(); void join();  }
@Test public void joinTest() throws InterruptedException { ExecutorPipeline<Integer, Integer> exec1 = new ExecutorPipeline<>(8, 100, true, integer -> { try { Thread.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); } return integer; }, Throwable::printStackTrace); final List<Integer> consumed = new ArrayList<>(); ExecutorPipeline<Integer, Void> exec2 = exec1.add(1, 100, consumed::add); int cnt = 1000; for (int i = 0; i < cnt; i++) { exec1.push(i); } exec1.join(); Assert.assertEquals(cnt, consumed.size()); }