### Question:
Parallelizer { public <T> void parallelize(Stream<? extends T> inputs, Consumer<? super T> consumer) throws InterruptedException { parallelize(forAll(inputs, consumer)); } Parallelizer(ExecutorService executor, int maxInFlight); void parallelize(Stream<? extends T> inputs, Consumer<? super T> consumer); void parallelize(Iterator<? extends T> inputs, Consumer<? super T> consumer); void parallelize(
      Stream<? extends T> inputs, Consumer<? super T> consumer,
      Duration heartbeatTimeout); void parallelize(
      Stream<? extends T> inputs, Consumer<? super T> consumer,
      long heartbeatTimeout, TimeUnit timeUnit); void parallelize(
      Iterator<? extends T> inputs, Consumer<? super T> consumer,
      Duration heartbeatTimeout); void parallelize(
      Iterator<? extends T> inputs, Consumer<? super T> consumer,
      long heartbeatTimeout, TimeUnit timeUnit); void parallelizeUninterruptibly(
      Stream<? extends T> inputs, Consumer<? super T> consumer); void parallelizeUninterruptibly(
      Iterator<? extends T> inputs, Consumer<? super T> consumer); void parallelize(Stream<? extends Runnable> tasks); void parallelize(
      Stream<? extends Runnable> tasks, Duration heartbeatTimeout); void parallelize(
      Stream<? extends Runnable> tasks, long heartbeatTimeout, TimeUnit timeUnit); void parallelizeUninterruptibly(Stream<? extends Runnable> tasks);  }### Answer:
@Test public void testTaskExceptionDismissesPendingTasks() { maxInFlight = 2; UncheckedExecutionException exception = assertThrows( UncheckedExecutionException.class, () -> parallelize(Stream.of( () -> translateToString(1), () -> translateToString(1), () -> fail("foobar"), () -> fail("foobar"), () -> translateToString(5)))); assertThat(exception.getCause().getMessage()).contains("foobar"); assertThat(translated).containsEntry(1, "1"); assertThat(translated).doesNotContainKey(5); }
@Test public void testTaskExceptionCancelsInFlightTasks() throws InterruptedException { assumeFalse(threading == Threading.DIRECT); maxInFlight = 2; UncheckedExecutionException exception = assertThrows( UncheckedExecutionException.class, () -> parallelize(serialTasks( () -> translateToString(1), () -> blockFor(2), () -> fail("foobar"), () -> translateToString(4)))); assertThat(exception.getCause().getMessage()).contains("foobar"); shutdownAndAssertInterruptedKeys().containsExactly(2); assertThat(translated).containsEntry(1, "1"); assertThat(translated).doesNotContainKey(4); }
@Test public void testSubmissionTimeoutCancelsInFlightTasks() throws InterruptedException { assumeFalse(threading == Threading.DIRECT); assumeTrue(mode == Mode.INTERRUPTIBLY); maxInFlight = 2; timeout = Duration.ofMillis(1); assertThrows( TimeoutException.class, () -> parallelize(serialTasks( () -> blockFor(1), () -> blockFor(2), () -> translateToString(3)))); shutdownAndAssertInterruptedKeys().containsExactly(1, 2); assertThat(translated).doesNotContainKey(3); }
@Test public void testAwaitTimeoutCancelsInFlightTasks() throws InterruptedException { assumeFalse(threading == Threading.DIRECT); assumeTrue(mode == Mode.INTERRUPTIBLY); maxInFlight = 2; timeout = Duration.ofMillis(1); assertThrows( TimeoutException.class, () -> parallelize(serialTasks( () -> blockFor(1), () -> blockFor(2)))); shutdownAndAssertInterruptedKeys().contains(1); }
@Test public void testErrorPropagated() { Error error = new Error(); UncheckedExecutionException exception = assertThrows( UncheckedExecutionException.class, () -> parallelize(Stream.of(() -> raise(error)))); assertThat(exception.getCause()).isSameAs(error); }
@Test public void testExceptionPropagated() { RuntimeException exception = new RuntimeException(); UncheckedExecutionException caught = assertThrows( UncheckedExecutionException.class, () -> parallelize(Stream.of(() -> raise(exception)))); assertThat(caught.getCause()).isSameAs(exception); }