JAVA-6390: Move kafka articles from libraries-6 to new module

apache-kafka

Author: freelansam

libraries-6/src/main/java/com/baeldung/kafka/TransactionalMessageProducer.java renamed to apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/TransactionalMessageProducer.java

@@ -1,4 +1,4 @@
-package com.baeldung.kafka;
+package com.baeldung.kafka.exactlyonce;
 
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -24,16 +24,16 @@ public static void main(String[] args) {
 
         producer.initTransactions();
 
-        try{
+        try {
 
             producer.beginTransaction();
 
-            Stream.of(DATA_MESSAGE_1, DATA_MESSAGE_2).forEach(s -> producer.send(
-                    new ProducerRecord<String, String>("input", null, s)));
+            Stream.of(DATA_MESSAGE_1, DATA_MESSAGE_2)
+                .forEach(s -> producer.send(new ProducerRecord<String, String>("input", null, s)));
 
             producer.commitTransaction();
 
-        }catch (KafkaException e){
+        } catch (KafkaException e) {
 
             producer.abortTransaction();
 

libraries-6/src/main/java/com/baeldung/kafka/TransactionalWordCount.java renamed to apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/TransactionalWordCount.java

@@ -1,4 +1,4 @@
-package com.baeldung.kafka;
+package com.baeldung.kafka.exactlyonce;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -43,10 +43,11 @@ public static void main(String[] args) {
                 ConsumerRecords<String, String> records = consumer.poll(ofSeconds(60));
 
                 Map<String, Integer> wordCountMap = records.records(new TopicPartition(INPUT_TOPIC, 0))
-                        .stream()
-                        .flatMap(record -> Stream.of(record.value().split(" ")))
-                        .map(word -> Tuple.of(word, 1))
-                        .collect(Collectors.toMap(tuple -> tuple.getKey(), t1 -> t1.getValue(), (v1, v2) -> v1 + v2));
+                    .stream()
+                    .flatMap(record -> Stream.of(record.value()
+                        .split(" ")))
+                    .map(word -> Tuple.of(word, 1))
+                    .collect(Collectors.toMap(tuple -> tuple.getKey(), t1 -> t1.getValue(), (v1, v2) -> v1 + v2));
 
                 producer.beginTransaction();
 
@@ -56,7 +57,8 @@ public static void main(String[] args) {
 
                 for (TopicPartition partition : records.partitions()) {
                     List<ConsumerRecord<String, String>> partitionedRecords = records.records(partition);
-                    long offset = partitionedRecords.get(partitionedRecords.size() - 1).offset();
+                    long offset = partitionedRecords.get(partitionedRecords.size() - 1)
+                        .offset();
 
                     offsetsToCommit.put(partition, new OffsetAndMetadata(offset + 1));
                 }
@@ -72,7 +74,6 @@ public static void main(String[] args) {
 
         }
 
-
     }
 
     private static KafkaConsumer<String, String> createKafkaConsumer() {

libraries-6/src/main/java/com/baeldung/kafka/Tuple.java renamed to apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/Tuple.java

@@ -1,4 +1,4 @@
-package com.baeldung.kafka;
+package com.baeldung.kafka.exactlyonce;
 
 public class Tuple {
 
@@ -10,8 +10,8 @@ private Tuple(String key, Integer value) {
         this.value = value;
     }
 
-    public static Tuple of(String key, Integer value){
-        return new Tuple(key,value);
+    public static Tuple of(String key, Integer value) {
+        return new Tuple(key, value);
     }
 
     public String getKey() {