KafkaAppender implements LogAppender { KafkaProducer<byte[], byte[]> createProducer(KafkaURI uri) { var watch = new StopWatch(); try { Map<String, Object> config = Map.of(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, uri.bootstrapURIs, ProducerConfig.ACKS_CONFIG, "0", ProducerConfig.CLIENT_ID_CONFIG, "log-forwarder", ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.SNAPPY.name, ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 60 * 1000, ProducerConfig.LINGER_MS_CONFIG, 50L, ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 500L, ProducerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, 5L * 1000, ProducerConfig.MAX_BLOCK_MS_CONFIG, 30L * 1000); var serializer = new ByteArraySerializer(); var producer = new KafkaProducer<>(config, serializer, serializer); producerMetrics.set(producer.metrics()); return producer; } finally { logger.info("create kafka log producer, uri={}, elapsed={}", uri, watch.elapsed()); } } KafkaAppender(KafkaURI uri); @Override void append(ActionLogMessage message); @Override void append(StatMessage message); void start(); void stop(long timeoutInMs); final ProducerMetrics producerMetrics; }
@Test void createProducer() { KafkaProducer<byte[], byte[]> producer = appender.createProducer(new KafkaURI("localhost")); assertThat(producer).isNotNull(); producer.close(Duration.ZERO); }