[#noissue] Cleanup Kafka Configuration

pinot/pinot-kafka/pom.xml

@@ -47,6 +47,18 @@
             <artifactId>spring-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.navercorp.pinpoint</groupId>
+            <artifactId>pinpoint-commons-server</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-test</artifactId>
+        </dependency>
 
     </dependencies>
 

pinot/pinot-kafka/src/main/java/com/navercorp/pinpoint/pinot/kafka/KafkaConfiguration.java

@@ -1,66 +1,60 @@
 package com.navercorp.pinpoint.pinot.kafka;
 
-import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.kafka.DefaultKafkaProducerFactoryCustomizer;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.PropertySource;
-import org.springframework.core.env.Environment;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.ProducerFactory;
 
-import java.util.HashMap;
 import java.util.Map;
-import java.util.function.Consumer;
 
 @Configuration
-@PropertySource("classpath:/profiles/${pinpoint.profiles.active:release}/kafka-producer-factory.properties")
+@PropertySource(value = {
+        "classpath:/kafka-root.properties",
+        "classpath:/profiles/${pinpoint.profiles.active:release}/kafka-producer-factory.properties"
+})
+@EnableConfigurationProperties
 public class KafkaConfiguration {
 
     private final Logger logger = LogManager.getLogger(KafkaConfiguration.class);
 
-    public Map<String, Object> toConfig(KafkaProperties properties) {
-        Map<String, Object> config = new HashMap<>();
-        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
-
-        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, properties.getKeySerializer());
-        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, properties.getValueSerializer());
-
-        config.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, properties.getPartitionerClass());
-        config.put(ProducerConfig.ACKS_CONFIG, properties.getAcks());
-        config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, properties.getCompressionType());
-
-        return config;
+    public KafkaConfiguration() {
+        logger.info("Install {}", KafkaConfiguration.class);
     }
 
     @Bean
-    public ProducerFactory kafkaProducerFactory(KafkaProperties properties) {
-        logger.info("kafka {}:{}", ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
-        logger.debug("kafka config:{}", properties);
+    public ProducerFactory<?, ?> kafkaProducerFactory(
+            ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers) {
 
-        Map<String, Object> config = toConfig(properties);
-        return new DefaultKafkaProducerFactory<>(config);
-    }
+        KafkaProperties properties = kafkaProperties();
+        // ref : KafkaAutoConfiguration
+        Map<String, Object> producerProperties = properties.buildProducerProperties();
+        logger.info("ProducerProperties:{}", producerProperties);
 
-    @Bean
-    public KafkaProperties kafkaProperties(Environment env) {
-        KafkaProperties properties = new KafkaProperties();
+        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(producerProperties);
 
-        bindProperties(env, "pinpoint.metric.kafka.bootstrap.servers", properties::setBootstrapServers);
-        bindProperties(env, "pinpoint.metric.kafka.key.serializer", properties::setKeySerializer);
-        bindProperties(env, "pinpoint.metric.kafka.value.serializer", properties::setValueSerializer);
-        bindProperties(env, "pinpoint.metric.kafka.acks", properties::setAcks);
-        bindProperties(env, "pinpoint.metric.kafka.compressionType", properties::setCompressionType);
+        KafkaProperties.Producer producer = properties.getProducer();
+        String transactionIdPrefix = producer.getTransactionIdPrefix();
+        if (transactionIdPrefix != null) {
+            factory.setTransactionIdPrefix(transactionIdPrefix);
+        }
 
-        return properties;
+        customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
+        return factory;
     }
 
-    private void bindProperties(Environment env, String key, Consumer<String> consumer) {
-        String value = env.getProperty(key);
-        if (value != null) {
-            consumer.accept(value);
-        }
+    @Bean
+    @ConfigurationProperties(prefix = "pinpoint.metric.kafka")
+    public KafkaProperties kafkaProperties() {
+        return new KafkaProperties();
     }
 
+
 }

pinot/pinot-kafka/src/main/resources/kafka-root.properties

@@ -0,0 +1,10 @@
+pinpoint.metric.kafka.bootstrapServers=----common-metric-kafka-address----
+pinpoint.metric.kafka.clientId=pinpoint-pinot-kafka
+
+pinpoint.metric.kafka.producer.keySerializer=org.apache.kafka.common.serialization.StringSerializer
+pinpoint.metric.kafka.producer.valueSerializer=org.springframework.kafka.support.serializer.JsonSerializer
+pinpoint.metric.kafka.producer.compressionType=zstd
+pinpoint.metric.kafka.producer.acks=1
+pinpoint.metric.kafka.producer.properties.partitioner.class=org.apache.kafka.clients.producer.internals.DefaultPartitioner
+
+

pinot/pinot-kafka/src/main/resources/profiles/local/kafka-producer-factory.properties

@@ -1,5 +1 @@
-pinpoint.metric.kafka.bootstrap.servers=----local-metric-kafka-address----
-#pinpoint.metric.kafka.key.serializer=
-#pinpoint.metric.kafka.value.serializer=
-#pinpoint.metric.kafka.acks=1
-#pinpoint.metric.compressionType=zstd
+pinpoint.metric.kafka.bootstrapServers=----local-metric-kafka-address----

pinot/pinot-kafka/src/main/resources/profiles/release/kafka-producer-factory.properties

@@ -1,5 +1 @@
-pinpoint.metric.kafka.bootstrap.servers=----release-metric-kafka-address----
-#pinpoint.metric.kafka.key.serializer=
-#pinpoint.metric.kafka.value.serializer=
-#pinpoint.metric.kafka.acks=1
-#pinpoint.metric.compressionType=zstd
+pinpoint.metric.kafka.bootstrapServers=----release-metric-kafka-address----

pinot/pinot-kafka/src/test/java/com/navercorp/pinpoint/pinot/kafka/KafkaConfigurationTest.java

@@ -1,7 +1,7 @@
 package com.navercorp.pinpoint.pinot.kafka;
 
 import org.apache.kafka.clients.producer.ProducerConfig;
-import org.junit.jupiter.api.Assertions;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -10,18 +10,22 @@
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
+import java.util.List;
+
 /**
  * @author Woonduk Kang(emeroad)
  */
 @ExtendWith(SpringExtension.class)
 @ContextConfiguration(classes = KafkaConfiguration.class )
-@TestPropertySource(properties = "pinpoint.metric.kafka.bootstrap.servers=test-kafka-bootstrap")
+@TestPropertySource(properties = "pinpoint.metric.kafka.bootstrapServers=test-kafka-bootstrap")
 class KafkaConfigurationTest {
     @Autowired
     ProducerFactory producerFactory;
 
     @Test
     void test() {
-        Assertions.assertEquals("test-kafka-bootstrap", producerFactory.getConfigurationProperties().get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
+        List<String> actual = (List) producerFactory.getConfigurationProperties().get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
+        Assertions.assertThat(actual)
+                .containsExactly("test-kafka-bootstrap");
     }
 }