### Question:
Part3MultithreadingParallelization { @Complexity(HARD) public static Publisher<String> paralellizeLongRunningWorkOnUnboundedAmountOfThread(Flux<Callable<String>> streamOfLongRunningSources) { throw new RuntimeException("Not implemented"); }  @Complexity(EASY) static Publisher<String> publishOnParallelThreadScheduler(Flux<String> source); @Complexity(EASY) static Publisher<String> subscribeOnSingleThreadScheduler(Callable<String> blockingCall); @Complexity(EASY) static ParallelFlux<String> paralellizeWorkOnDifferentThreads(Flux<String> source); @Complexity(HARD) static Publisher<String> paralellizeLongRunningWorkOnUnboundedAmountOfThread(Flux<Callable<String>> streamOfLongRunningSources);  }### Answer:
@Test public void paralellizeLongRunningWorkOnUnboundedAmountOfThreadTest() { Thread[] threads = new Thread[3]; StepVerifier .create(paralellizeLongRunningWorkOnUnboundedAmountOfThread(Flux.<Callable<String>>just( () -> { threads[0] = Thread.currentThread(); Thread.sleep(300); return "Hello"; }, () -> { threads[1] = Thread.currentThread(); Thread.sleep(300); return "Hello"; }, () -> { threads[2] = Thread.currentThread(); Thread.sleep(300); return "Hello"; } ).repeat(20))) .expectSubscription() .expectNext("Hello", "Hello", "Hello") .expectNextCount(60) .expectComplete() .verify(Duration.ofMillis(600)); Assert.assertTrue( "Expected execution on different Threads", !threads[0].equals(threads[1]) ); Assert.assertTrue( "Expected execution on different Threads", !threads[1].equals(threads[2]) ); Assert.assertTrue( "Expected execution on different Threads", !threads[0].equals(threads[2]) ); }