KafkaAppender implements LogAppender { public void stop(long timeoutInMs) { logger.info("stop log forwarder"); stop = true; logForwarderThread.interrupt(); if (producer != null) { for (ProducerRecord<byte[], byte[]> record : records) { producer.send(record); } producer.close(Duration.ofMillis(timeoutInMs)); } } KafkaAppender(KafkaURI uri); @Override void append(ActionLogMessage message); @Override void append(StatMessage message); void start(); void stop(long timeoutInMs); final ProducerMetrics producerMetrics; }
@Test void stop() { appender.stop(-1); }