#15 kafka : producer - simple producer

kafka/kafka-producer/src/main/java/KafkaConfig.java

@@ -0,0 +1,18 @@
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.util.Properties;
+
+public class KafkaConfig {
+    protected final static String TOPIC_NAME = "test";
+    protected final static String BOOTSTRAP_SERVERS = "localhost:9092";
+
+    protected static KafkaProducer<String, String> setup() {
+        Properties configs = new Properties();
+        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfig.BOOTSTRAP_SERVERS);
+        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        return new KafkaProducer<>(configs);
+    }
+}

kafka/kafka-producer/src/main/java/SimpleProducer.java

@@ -0,0 +1,23 @@
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+
+public class SimpleProducer {
+    private final static Logger logger = LoggerFactory.getLogger(SimpleProducer.class);
+
+    public static void main(String[] args) {
+        KafkaProducer<String, String> producer = KafkaConfig.setup();
+
+        String messageValue = "testMessage";
+        ProducerRecord<String, String> record = new ProducerRecord<>(KafkaConfig.TOPIC_NAME, messageValue);
+        producer.send(record);
+        logger.info("{}", record);
+        producer.flush();
+        producer.close();
+    }
+}