### Question:
DynamicParallelOracle extends AbstractDynamicBatchProcessor<Query<I, D>, MembershipOracle<I, D>> implements ParallelOracle<I, D> { @Override public void processQueries(Collection<? extends Query<I, D>> queries) { processBatch(queries); } DynamicParallelOracle(Supplier<? extends MembershipOracle<I, D>> oracleSupplier,
                                 @NonNegative int batchSize,
                                 ExecutorService executor); @Override void processQueries(Collection<? extends Query<I, D>> queries);  }### Answer:
@Test(dataProvider = "policies", dataProviderClass = Utils.class, timeOut = 2000) public void testThreadCreation(PoolPolicy poolPolicy) { final List<AnswerOnceQuery<Void>> queries = createQueries(10); final int expectedThreads = queries.size(); final CountDownLatch latch = new CountDownLatch(expectedThreads); final NullOracle[] oracles = new NullOracle[expectedThreads]; for (int i = 0; i < expectedThreads; i++) { oracles[i] = new NullOracle() { @Override public void processQueries(Collection<? extends Query<Void, Void>> queries) { try { latch.countDown(); latch.await(); } catch (InterruptedException e) { throw new IllegalStateException(e); } super.processQueries(queries); } }; } final ParallelOracle<Void, Void> oracle = ParallelOracleBuilders.newDynamicParallelOracle(oracles[0], Arrays.copyOfRange( oracles, 1, oracles.length)) .withBatchSize(1) .withPoolSize(oracles.length) .withPoolPolicy(poolPolicy) .create(); try { oracle.processQueries(queries); } finally { oracle.shutdown(); } }
@Test(dataProvider = "policies", dataProviderClass = Utils.class, timeOut = 2000) public void testThreadScheduling(PoolPolicy poolPolicy) { final List<AnswerOnceQuery<Void>> queries = createQueries(10); final CountDownLatch latch = new CountDownLatch(queries.size() - 1); final NullOracle awaitingOracle = new NullOracle() { @Override public void processQueries(Collection<? extends Query<Void, Void>> queries) { try { latch.await(); } catch (InterruptedException e) { throw new IllegalStateException(e); } super.processQueries(queries); } }; final NullOracle countDownOracle = new NullOracle() { @Override public void processQueries(Collection<? extends Query<Void, Void>> queries) { latch.countDown(); super.processQueries(queries); } }; final ParallelOracle<Void, Void> oracle = ParallelOracleBuilders.newDynamicParallelOracle(awaitingOracle, countDownOracle) .withPoolSize(2) .withPoolPolicy(poolPolicy) .create(); try { oracle.processQueries(queries); } finally { oracle.shutdown(); } }