MessageProducer { public void close(long timeoutInMs) { if (producer != null) { logger.info("close kafka producer, uri={}, name={}", uri, name); producer.flush(); producer.close(Duration.ofMillis(timeoutInMs)); } } MessageProducer(KafkaURI uri, String name); void send(ProducerRecord<byte[], byte[]> record); void tryCreateProducer(); void close(long timeoutInMs); final ProducerMetrics producerMetrics; }
@Test void close() { producer.close(-1); }