KafkaAccessLogWriter implements AccessLogWriter { @Override public CompletableFuture<Void> shutdown() { return CompletableFuture.runAsync(producer::close); } KafkaAccessLogWriter(Producer<K, V> producer, String topic,
                                Function<? super RequestLog, ? extends V> valueExtractor);  KafkaAccessLogWriter(Producer<K, V> producer, String topic,
                                Function<? super RequestLog, ? extends K> keyExtractor,
                                Function<? super RequestLog, ? extends V> valueExtractor); @Override void log(RequestLog log); @Override CompletableFuture<Void> shutdown();  }
@Test void closeProducerWhenRequested() { final KafkaAccessLogWriter<String, String> service = new KafkaAccessLogWriter<>(producer, TOPIC_NAME, log -> ""); service.shutdown().join(); verify(producer, times(1)).close(); }