### Question:
Part4ExtraBackpressure_Optional { @Optional @Complexity(MEDIUM) public static void dynamicDemand(Flux<String> source, CountDownLatch countDownOnComplete) { throw new RuntimeException("Not implemented"); }  @Complexity(MEDIUM) static Publisher<String> handleBackpressureWithBuffering(StringEventPublisher stringEventPublisher); @Optional @Complexity(MEDIUM) static void dynamicDemand(Flux<String> source, CountDownLatch countDownOnComplete);  }### Answer:
@Test public void dynamicDemandTest() throws InterruptedException { int size = 100000; long requests = (long) Math.ceil((Math.log(size) / Math.log(2) + 1e-10)); CountDownLatch latch = new CountDownLatch(1); AtomicInteger iterations = new AtomicInteger(); dynamicDemand( Flux.range(0, size) .map(String::valueOf) .publishOn(Schedulers.single()) .doOnRequest(r -> iterations.incrementAndGet()), latch ); latch.await(5, TimeUnit.SECONDS); assertEquals(requests, iterations.get()); }