Merge branch 'eugenp:master' into master

algorithms-miscellaneous-2/src/test/java/com/baeldung/algorithms/astar/RouteFinder.java

@@ -8,6 +8,9 @@ import java.util.PriorityQueue;
 import java.util.Queue;
 import java.util.stream.Collectors;
 
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
 public class RouteFinder<T extends GraphNode> {
     private final Graph<T> graph;
     private final Scorer<T> nextNodeScorer;
@@ -28,11 +31,11 @@ public class RouteFinder<T extends GraphNode> {
         openSet.add(start);
 
         while (!openSet.isEmpty()) {
-            System.out.println("Open Set contains: " + openSet.stream().map(RouteNode::getCurrent).collect(Collectors.toSet()));
+            log.debug("Open Set contains: " + openSet.stream().map(RouteNode::getCurrent).collect(Collectors.toSet()));
             RouteNode<T> next = openSet.poll();
-            System.out.println("Looking at node: " + next);
+            log.debug("Looking at node: " + next);
             if (next.getCurrent().equals(to)) {
-                System.out.println("Found our destination!");
+                log.debug("Found our destination!");
 
                 List<T> route = new ArrayList<>();
                 RouteNode<T> current = next;
@@ -41,7 +44,7 @@ public class RouteFinder<T extends GraphNode> {
                     current = allNodes.get(current.getPrevious());
                 } while (current != null);
 
-                System.out.println("Route: " + route);
+                log.debug("Route: " + route);
                 return route;
             }
 
@@ -55,7 +58,7 @@ public class RouteFinder<T extends GraphNode> {
                     nextNode.setRouteScore(newScore);
                     nextNode.setEstimatedScore(newScore + targetScorer.computeCost(connection, to));
                     openSet.add(nextNode);
-                    System.out.println("Found a better route to node: " + nextNode);
+                    log.debug("Found a better route to node: " + nextNode);
                 }
             });
         }

algorithms-miscellaneous-2/src/test/java/com/baeldung/algorithms/astar/underground/RouteFinderIntegrationTest.java

@@ -1,5 +1,7 @@
 package com.baeldung.algorithms.astar.underground;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -10,9 +12,13 @@ import java.util.stream.Stream;
 
 import com.baeldung.algorithms.astar.Graph;
 import com.baeldung.algorithms.astar.RouteFinder;
+
+import lombok.extern.slf4j.Slf4j;
+
 import org.junit.Before;
 import org.junit.Test;
 
+@Slf4j
 public class RouteFinderIntegrationTest {
 
     private Graph<Station> underground;
@@ -637,7 +643,8 @@ public class RouteFinderIntegrationTest {
     @Test
     public void findRoute() {
         List<Station> route = routeFinder.findRoute(underground.getNode("74"), underground.getNode("7"));
+        assertThat(route).size().isPositive();
 
-        System.out.println(route.stream().map(Station::getName).collect(Collectors.toList()));
+        route.stream().map(Station::getName).collect(Collectors.toList()).forEach(station -> log.debug(station));
     }
 }

apache-kafka/README.md

@@ -0,0 +1,18 @@
+## Apache Kafka
+
+This module contains articles about Apache Kafka.
+
+### Relevant articles
+- [Kafka Streams vs Kafka Consumer](https://www.baeldung.com/java-kafka-streams-vs-kafka-consumer)
+- [Kafka Topic Creation Using Java](https://www.baeldung.com/kafka-topic-creation)
+- [Using Kafka MockConsumer](https://www.baeldung.com/kafka-mockconsumer)
+- [Using Kafka MockProducer](https://www.baeldung.com/kafka-mockproducer)
+- [Introduction to KafkaStreams in Java](https://www.baeldung.com/java-kafka-streams)
+- [Introduction to Kafka Connectors](https://www.baeldung.com/kafka-connectors-guide)
+- [Kafka Connect Example with MQTT and MongoDB](https://www.baeldung.com/kafka-connect-mqtt-mongodb)
+- [Building a Data Pipeline with Flink and Kafka](https://www.baeldung.com/kafka-flink-data-pipeline)
+- [Exactly Once Processing in Kafka with Java](https://www.baeldung.com/kafka-exactly-once)
+
+
+##### Building the project
+You can build the project from the command line using: *mvn clean install*, or in an IDE.

apache-kafka/pom.xml

@@ -0,0 +1,180 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>apache-kafka</artifactId>
+    <name>apache-kafka</name>
+
+    <parent>
+        <groupId>com.baeldung</groupId>
+        <artifactId>parent-modules</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>${kafka.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams</artifactId>
+            <version>${kafka.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${org.slf4j.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>${org.slf4j.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_2.11</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>commons-logging</artifactId>
+                    <groupId>commons-logging</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-java</artifactId>
+            <version>${flink.version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>commons-logging</artifactId>
+                    <groupId>commons-logging</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils_2.11</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${guava.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility-proxy</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <version>${testcontainers-kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>${testcontainers-jupiter.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_2.11</artifactId>
+            <version>${org.apache.spark.spark-core.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_2.11</artifactId>
+            <version>${org.apache.spark.spark-core.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-graphx_2.11</artifactId>
+            <version>${org.apache.spark.spark-core.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-streaming_2.11</artifactId>
+            <version>${org.apache.spark.spark-core.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-mllib_2.11</artifactId>
+            <version>${org.apache.spark.spark-core.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
+            <version>${org.apache.spark.spark-core.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.datastax.spark</groupId>
+            <artifactId>spark-cassandra-connector_2.11</artifactId>
+            <version>${com.datastax.spark.spark-cassandra-connector.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.datastax.spark</groupId>
+            <artifactId>spark-cassandra-connector-java_2.11</artifactId>
+            <version>${com.datastax.spark.spark-cassandra-connector-java.version}</version>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <assertj.version>3.6.2</assertj.version>
+        <kafka.version>2.8.0</kafka.version>
+        <testcontainers-kafka.version>1.15.3</testcontainers-kafka.version>
+        <testcontainers-jupiter.version>1.15.3</testcontainers-jupiter.version>
+        <flink.version>1.5.0</flink.version>
+        <awaitility.version>3.0.0</awaitility.version>
+        <guava.version>29.0-jre</guava.version>
+        <org.apache.spark.spark-core.version>2.4.8</org.apache.spark.spark-core.version>
+        <graphframes.version>0.8.1-spark3.0-s_2.12</graphframes.version>
+        <com.datastax.spark.spark-cassandra-connector.version>2.5.2</com.datastax.spark.spark-cassandra-connector.version>
+        <com.datastax.spark.spark-cassandra-connector-java.version>1.6.0-M1</com.datastax.spark.spark-cassandra-connector-java.version>
+    </properties>
+
+</project>

apache-kafka/src/main/java/com/baeldung/flink/FlinkDataPipeline.java

@@ -0,0 +1,70 @@
+package com.baeldung.flink;
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.model.InputMessage;
+import com.baeldung.flink.operator.BackupAggregator;
+import com.baeldung.flink.operator.InputMessageTimestampAssigner;
+import com.baeldung.flink.operator.WordsCapitalizer;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
+
+import static com.baeldung.flink.connector.Consumers.*;
+import static com.baeldung.flink.connector.Producers.*;
+
+public class FlinkDataPipeline {
+
+    public static void capitalize() throws Exception {
+        String inputTopic = "flink_input";
+        String outputTopic = "flink_output";
+        String consumerGroup = "baeldung";
+        String address = "localhost:9092";
+
+        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        FlinkKafkaConsumer011<String> flinkKafkaConsumer = createStringConsumerForTopic(inputTopic, address, consumerGroup);
+        flinkKafkaConsumer.setStartFromEarliest();
+
+        DataStream<String> stringInputStream = environment.addSource(flinkKafkaConsumer);
+
+        FlinkKafkaProducer011<String> flinkKafkaProducer = createStringProducer(outputTopic, address);
+
+        stringInputStream.map(new WordsCapitalizer())
+            .addSink(flinkKafkaProducer);
+
+        environment.execute();
+    }
+
+    public static void createBackup() throws Exception {
+        String inputTopic = "flink_input";
+        String outputTopic = "flink_output";
+        String consumerGroup = "baeldung";
+        String kafkaAddress = "localhost:9092";
+
+        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+        FlinkKafkaConsumer011<InputMessage> flinkKafkaConsumer = createInputMessageConsumer(inputTopic, kafkaAddress, consumerGroup);
+        flinkKafkaConsumer.setStartFromEarliest();
+
+        flinkKafkaConsumer.assignTimestampsAndWatermarks(new InputMessageTimestampAssigner());
+        FlinkKafkaProducer011<Backup> flinkKafkaProducer = createBackupProducer(outputTopic, kafkaAddress);
+
+        DataStream<InputMessage> inputMessagesStream = environment.addSource(flinkKafkaConsumer);
+
+        inputMessagesStream.timeWindowAll(Time.hours(24))
+            .aggregate(new BackupAggregator())
+            .addSink(flinkKafkaProducer);
+
+        environment.execute();
+    }
+
+    public static void main(String[] args) throws Exception {
+        createBackup();
+    }
+
+}

apache-kafka/src/main/java/com/baeldung/flink/connector/Consumers.java

@@ -9,23 +9,20 @@ import java.util.Properties;
 
 public class Consumers {
 
-public static FlinkKafkaConsumer011<String> createStringConsumerForTopic(
-  String topic, String kafkaAddress, String kafkaGroup ) {
-    Properties props = new Properties();
-    props.setProperty("bootstrap.servers", kafkaAddress);
-    props.setProperty("group.id",kafkaGroup);
-    FlinkKafkaConsumer011<String> consumer =
-      new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(),props);
+    public static FlinkKafkaConsumer011<String> createStringConsumerForTopic(String topic, String kafkaAddress, String kafkaGroup) {
+        Properties props = new Properties();
+        props.setProperty("bootstrap.servers", kafkaAddress);
+        props.setProperty("group.id", kafkaGroup);
+        FlinkKafkaConsumer011<String> consumer = new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(), props);
 
-    return consumer;
-}
+        return consumer;
+    }
 
-    public static FlinkKafkaConsumer011<InputMessage> createInputMessageConsumer(String topic, String kafkaAddress, String kafkaGroup ) {
+    public static FlinkKafkaConsumer011<InputMessage> createInputMessageConsumer(String topic, String kafkaAddress, String kafkaGroup) {
         Properties properties = new Properties();
         properties.setProperty("bootstrap.servers", kafkaAddress);
-        properties.setProperty("group.id",kafkaGroup);
-        FlinkKafkaConsumer011<InputMessage> consumer = new FlinkKafkaConsumer011<InputMessage>(
-                topic, new InputMessageDeserializationSchema(),properties);
+        properties.setProperty("group.id", kafkaGroup);
+        FlinkKafkaConsumer011<InputMessage> consumer = new FlinkKafkaConsumer011<InputMessage>(topic, new InputMessageDeserializationSchema(), properties);
 
         return consumer;
     }

apache-kafka/src/main/java/com/baeldung/flink/model/InputMessage.java

@@ -18,6 +18,7 @@ public class InputMessage {
     public String getSender() {
         return sender;
     }
+
     public void setSender(String sender) {
         this.sender = sender;
     }
@@ -55,12 +56,14 @@ public class InputMessage {
 
     @Override
     public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
+        if (this == o)
+            return true;
+        if (o == null || getClass() != o.getClass())
+            return false;
         InputMessage message1 = (InputMessage) o;
-        return Objects.equal(sender, message1.sender) &&
-                Objects.equal(recipient, message1.recipient) &&
-                Objects.equal(sentAt, message1.sentAt) &&
+        return Objects.equal(sender, message1.sender) && 
+                Objects.equal(recipient, message1.recipient) && 
+                Objects.equal(sentAt, message1.sentAt) && 
                 Objects.equal(message, message1.message);
     }
 

apache-kafka/src/main/java/com/baeldung/flink/operator/BackupAggregator.java

@@ -0,0 +1,34 @@
+package com.baeldung.flink.operator;
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.model.InputMessage;
+import org.apache.flink.api.common.functions.AggregateFunction;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BackupAggregator implements AggregateFunction<InputMessage, List<InputMessage>, Backup> {
+    @Override
+    public List<InputMessage> createAccumulator() {
+        return new ArrayList<>();
+    }
+
+    @Override
+    public List<InputMessage> add(InputMessage inputMessage, List<InputMessage> inputMessages) {
+        inputMessages.add(inputMessage);
+        return inputMessages;
+    }
+
+    @Override
+    public Backup getResult(List<InputMessage> inputMessages) {
+        Backup backup = new Backup(inputMessages, LocalDateTime.now());
+        return backup;
+    }
+
+    @Override
+    public List<InputMessage> merge(List<InputMessage> inputMessages, List<InputMessage> acc1) {
+        inputMessages.addAll(acc1);
+        return inputMessages;
+    }
+}

apache-kafka/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java

@@ -12,7 +12,9 @@ public class InputMessageTimestampAssigner implements AssignerWithPunctuatedWate
     @Override
     public long extractTimestamp(InputMessage element, long previousElementTimestamp) {
         ZoneId zoneId = ZoneId.systemDefault();
-        return element.getSentAt().atZone(zoneId).toEpochSecond() * 1000;
+        return element.getSentAt()
+            .atZone(zoneId)
+            .toEpochSecond() * 1000;
     }
 
     @Nullable

apache-kafka/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java

@@ -9,8 +9,7 @@ import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class BackupSerializationSchema
-  implements SerializationSchema<Backup> {
+public class BackupSerializationSchema implements SerializationSchema<Backup> {
 
     static ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
 
@@ -18,7 +17,7 @@ public class BackupSerializationSchema
 
     @Override
     public byte[] serialize(Backup backupMessage) {
-        if(objectMapper == null) {
+        if (objectMapper == null) {
             objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
             objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
         }

apache-kafka/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java

@@ -8,12 +8,10 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
 
 import java.io.IOException;
 
-public class InputMessageDeserializationSchema implements
-      DeserializationSchema<InputMessage> {
+public class InputMessageDeserializationSchema implements DeserializationSchema<InputMessage> {
 
     static ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
 
-
     @Override
     public InputMessage deserialize(byte[] bytes) throws IOException {
 

apache-kafka/src/main/java/com/baeldung/kafka/admin/KafkaTopicApplication.java

@@ -27,11 +27,11 @@ public class KafkaTopicApplication {
             short replicationFactor = 1;
             NewTopic newTopic = new NewTopic(topicName, partitions, replicationFactor);
 
-            CreateTopicsResult result = admin.createTopics(
-              Collections.singleton(newTopic));
+            CreateTopicsResult result = admin.createTopics(Collections.singleton(newTopic));
 
             // get the async result for the new topic creation
-            KafkaFuture<Void> future = result.values().get(topicName);
+            KafkaFuture<Void> future = result.values()
+                .get(topicName);
 
             // call get() to block until topic creation has completed or failed
             future.get();
@@ -47,15 +47,13 @@ public class KafkaTopicApplication {
             short replicationFactor = 1;
             NewTopic newTopic = new NewTopic(topicName, partitions, replicationFactor);
 
-            CreateTopicsOptions topicOptions = new CreateTopicsOptions()
-              .validateOnly(true)
-              .retryOnQuotaViolation(true);
+            CreateTopicsOptions topicOptions = new CreateTopicsOptions().validateOnly(true)
+                .retryOnQuotaViolation(true);
 
-            CreateTopicsResult result = admin.createTopics(
-              Collections.singleton(newTopic), topicOptions
-            );
+            CreateTopicsResult result = admin.createTopics(Collections.singleton(newTopic), topicOptions);
 
-            KafkaFuture<Void> future = result.values().get(topicName);
+            KafkaFuture<Void> future = result.values()
+                .get(topicName);
             future.get();
         }
     }
@@ -72,14 +70,12 @@ public class KafkaTopicApplication {
             Map<String, String> newTopicConfig = new HashMap<>();
             newTopicConfig.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
             newTopicConfig.put(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4");
-            NewTopic newTopic = new NewTopic(topicName, partitions, replicationFactor)
-              .configs(newTopicConfig);
+            NewTopic newTopic = new NewTopic(topicName, partitions, replicationFactor).configs(newTopicConfig);
 
-            CreateTopicsResult result = admin.createTopics(
-              Collections.singleton(newTopic)
-            );
+            CreateTopicsResult result = admin.createTopics(Collections.singleton(newTopic));
 
-            KafkaFuture<Void> future = result.values().get(topicName);
+            KafkaFuture<Void> future = result.values()
+                .get(topicName);
             future.get();
         }
     }

apache-kafka/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java

@@ -19,9 +19,7 @@ public class CountryPopulationConsumer {
     private java.util.function.Consumer<Throwable> exceptionConsumer;
     private java.util.function.Consumer<CountryPopulation> countryPopulationConsumer;
 
-    public CountryPopulationConsumer(
-            Consumer<String, Integer> consumer, java.util.function.Consumer<Throwable> exceptionConsumer,
-            java.util.function.Consumer<CountryPopulation> countryPopulationConsumer) {
+    public CountryPopulationConsumer(Consumer<String, Integer> consumer, java.util.function.Consumer<Throwable> exceptionConsumer, java.util.function.Consumer<CountryPopulation> countryPopulationConsumer) {
         this.consumer = consumer;
         this.exceptionConsumer = exceptionConsumer;
         this.countryPopulationConsumer = countryPopulationConsumer;

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 class TransactionalMessageProducer {
 
         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();
 

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 class TransactionalWordCount {
                 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 class TransactionalWordCount {
 
                 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 class TransactionalWordCount {
 
         }
 
-
     }
 
     private static KafkaConsumer<String, String> createKafkaConsumer() {

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 @@ public class Tuple {
         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() {

apache-kafka/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java

@@ -15,8 +15,7 @@ public class KafkaProducer {
     }
 
     public Future<RecordMetadata> send(String key, String value) {
-        ProducerRecord record = new ProducerRecord("topic_sports_news",
-                key, value);
+        ProducerRecord record = new ProducerRecord("topic_sports_news", key, value);
         return producer.send(record);
     }
 
@@ -36,5 +35,4 @@ public class KafkaProducer {
         producer.commitTransaction();
     }
 
-
 }