producer properties review

Author: hifly81

kafka-distributed-tracing/docker-compose-tracing.yml

@@ -48,4 +48,4 @@ services:
 
 volumes:
   db:
-    driver: local
+    driver: local

kafka-producer/src/main/java/org/hifly/kafka/demo/producer/KafkaConfig.java

@@ -1,14 +1,10 @@
 package org.hifly.kafka.demo.producer;
 
-import com.hortonworks.registries.schemaregistry.serdes.avro.kafka.KafkaAvroSerializer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringSerializer;
 
-import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.Paths;
 import java.util.Properties;
 
 public class KafkaConfig {
@@ -24,16 +20,17 @@ public class KafkaConfig {
 
 
     public static Properties loadConfig(final String configFile) throws IOException {
-        if (!Files.exists(Paths.get(configFile))) {
-            throw new IOException(configFile + " not found.");
-        }
         final Properties cfg = new Properties();
-        try (InputStream inputStream = new FileInputStream(configFile)) {
+
+        ClassLoader classLoader = KafkaConfig.class.getClassLoader();
+        InputStream inputStream = classLoader.getResourceAsStream(configFile);
+
+        if (inputStream == null) {
+            throw new IllegalArgumentException("file not found! " + configFile);
+        } else {
             cfg.load(inputStream);
         }
 
-        cfg.put("acks", "all");
-
         return cfg;
     }
     public static Properties stringProducer() {

kafka-producer/src/main/java/org/hifly/kafka/demo/producer/serializer/string/Runner.java

@@ -1,20 +1,20 @@
 package org.hifly.kafka.demo.producer.serializer.string;
 
-import org.hifly.kafka.demo.producer.ProducerCallback;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.hifly.kafka.demo.producer.KafkaConfig;
 import org.hifly.kafka.demo.producer.RecordMetadataUtil;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 
 public class Runner {
 
     public static void main (String [] args) throws Exception {
+        KafkaProducer<String, String> producer = new KafkaProducer<>(
+                KafkaConfig.loadConfig("produce-kip-714.properties"));
         StringProducer baseProducer = new StringProducer();
-        baseProducer.start();
+        baseProducer.start(producer);
         bunchOfSynchMessages("topic1", baseProducer);
-        baseProducer.start();
-        bunchOfFFMessages("topic1", baseProducer);
-        baseProducer.start();
-        bunchOfAsynchMessages("topic1", baseProducer);
     }
 
     public static void bunchOfSynchMessages(String topic, StringProducer baseProducer) {
@@ -26,15 +26,4 @@ public static void bunchOfSynchMessages(String topic, StringProducer baseProduce
         baseProducer.stop();
     }
 
-    public static void bunchOfFFMessages(String topic, StringProducer baseProducer) {
-        for (int i= 0; i < 10; i++ )
-             baseProducer.produceFireAndForget(new ProducerRecord<>(topic, Integer.toString(i)));
-        baseProducer.stop();
-    }
-
-    public static void bunchOfAsynchMessages(String topic, StringProducer baseProducer) {
-        for (int i= 0; i < 10; i++ )
-            baseProducer.produceAsync(new ProducerRecord<>(topic, Integer.toString(i)), new ProducerCallback());
-        baseProducer.stop();
-    }
 }

kafka-producer/src/main/resources/producer.properties

@@ -1,10 +1,14 @@
 bootstrap.servers=
+key.serializer=
+value.serializer
 security.protocol=
 sasl.kerberos.service.name=
 sasl.mechanism=
 sasl.jaas.config=
 ssl.truststore.location=
 ssl.truststore.password=
+client.id=
+enable.metrics.push=
 schema.registry.url=
 schema.registry.ssl.truststore.location=
 schema.registry.ssl.truststore.password=