KafkaAccessLogWriter implements AccessLogWriter { @Override public void log(RequestLog log) { final V value = valueExtractor.apply(log); if (value == null) { return; } final K key = keyExtractor.apply(log); final ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, value); producer.send(producerRecord, (metadata, exception) -> { if (exception != null) { logger.warn("Failed to send a record to Kafka: {}", producerRecord, exception); } }); } 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 withoutKeyExtractor() { final KafkaAccessLogWriter<String, String> service = new KafkaAccessLogWriter<>(producer, TOPIC_NAME, log -> log.requestHeaders().authority()); service.log(log); verify(producer, times(1)).send(captor.capture(), any(Callback.class)); final ProducerRecord<String, String> record = captor.getValue(); assertThat(record.key()).isNull(); assertThat(record.value()).isEqualTo("kawamuray"); }
@Test void withKeyExtractor() { final KafkaAccessLogWriter<String, String> service = new KafkaAccessLogWriter<>(producer, TOPIC_NAME, log -> log.context().decodedPath(), log -> log.requestHeaders().authority()); service.log(log); verify(producer, times(1)).send(captor.capture(), any(Callback.class)); final ProducerRecord<String, String> record = captor.getValue(); assertThat(record.key()).isEqualTo("/kyuto"); assertThat(record.value()).isEqualTo("kawamuray"); }