From 62eaeb4835c52fae266bd53d09bb032a9f4ba6af Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:02:31 +0530 Subject: [PATCH 01/12] JAVA-6390: Move kafka articles from libraries-data-3 to new module apache-kafka --- .../kafka/admin/KafkaTopicApplication.java | 28 +- .../KafkaTopicApplicationIntegrationTest.java | 0 .../kafkastreams/KafkaStreamsLiveTest.java | 77 +++++ .../kafka/streams/KafkaStreamsLiveTest.java | 279 ------------------ 4 files changed, 89 insertions(+), 295 deletions(-) rename {libraries-data-3 => apache-kafka}/src/main/java/com/baeldung/kafka/admin/KafkaTopicApplication.java (77%) rename {libraries-data-3 => apache-kafka}/src/test/java/com/baeldung/kafka/admin/KafkaTopicApplicationIntegrationTest.java (100%) create mode 100644 apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java delete mode 100644 libraries-data-3/src/test/java/com/baeldung/kafka/streams/KafkaStreamsLiveTest.java diff --git a/libraries-data-3/src/main/java/com/baeldung/kafka/admin/KafkaTopicApplication.java b/apache-kafka/src/main/java/com/baeldung/kafka/admin/KafkaTopicApplication.java similarity index 77% rename from libraries-data-3/src/main/java/com/baeldung/kafka/admin/KafkaTopicApplication.java rename to apache-kafka/src/main/java/com/baeldung/kafka/admin/KafkaTopicApplication.java index 0d74e27d4e..25d621166d 100644 --- a/libraries-data-3/src/main/java/com/baeldung/kafka/admin/KafkaTopicApplication.java +++ b/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 future = result.values().get(topicName); + KafkaFuture 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 future = result.values().get(topicName); + KafkaFuture future = result.values() + .get(topicName); future.get(); } } @@ -72,14 +70,12 @@ public class KafkaTopicApplication { Map 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 future = result.values().get(topicName); + KafkaFuture future = result.values() + .get(topicName); future.get(); } } diff --git a/libraries-data-3/src/test/java/com/baeldung/kafka/admin/KafkaTopicApplicationIntegrationTest.java b/apache-kafka/src/test/java/com/baeldung/kafka/admin/KafkaTopicApplicationIntegrationTest.java similarity index 100% rename from libraries-data-3/src/test/java/com/baeldung/kafka/admin/KafkaTopicApplicationIntegrationTest.java rename to apache-kafka/src/test/java/com/baeldung/kafka/admin/KafkaTopicApplicationIntegrationTest.java diff --git a/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java b/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java new file mode 100644 index 0000000000..0b66dd8fec --- /dev/null +++ b/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java @@ -0,0 +1,77 @@ +package com.baeldung.kafkastreams; + +import java.util.Arrays; +import java.util.Properties; +import java.util.regex.Pattern; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; +import org.apache.kafka.streams.kstream.Produced; +import org.junit.Ignore; +import org.junit.Test; + +public class KafkaStreamsLiveTest { + private String bootstrapServers = "localhost:9092"; + + @Test + @Ignore("it needs to have kafka broker running on local") + public void shouldTestKafkaStreams() throws InterruptedException { + //given + String inputTopic = "inputTopic"; + + Properties streamsConfiguration = new Properties(); + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-live-test"); + streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); + streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); + streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); + streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + // Use a temporary directory for storing state, which will be automatically removed after the test. + // streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath()); + + /* + * final StreamsBuilder builder = new StreamsBuilder(); + KStream textLines = builder.stream(wordCountTopic, + Consumed.with(Serdes.String(), Serdes.String())); + + KTable wordCounts = textLines + .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT) + .split("\\W+"))) + .groupBy((key, word) -> word) + .count(Materialized.> as("counts-store")); + */ + //when + final StreamsBuilder builder = new StreamsBuilder(); + KStream textLines = builder.stream(inputTopic); + Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS); + + KTable wordCounts = textLines + .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase()))) + .groupBy((key, word) -> word) + .count(); + + wordCounts.toStream().foreach((word, count) -> System.out.println("word: " + word + " -> " + count)); + + String outputTopic = "outputTopic"; + //final Serde stringSerde = Serdes.String(); + //final Serde longSerde = Serdes.Long(); + //wordCounts.toStream().to(stringSerde, longSerde, outputTopic); + + wordCounts.toStream().to("outputTopic", + Produced.with(Serdes.String(), Serdes.Long())); + + final Topology topology = builder.build(); + KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); + streams.start(); + + //then + Thread.sleep(30000); + streams.close(); + } +} diff --git a/libraries-data-3/src/test/java/com/baeldung/kafka/streams/KafkaStreamsLiveTest.java b/libraries-data-3/src/test/java/com/baeldung/kafka/streams/KafkaStreamsLiveTest.java deleted file mode 100644 index 0d4c0606e3..0000000000 --- a/libraries-data-3/src/test/java/com/baeldung/kafka/streams/KafkaStreamsLiveTest.java +++ /dev/null @@ -1,279 +0,0 @@ -package com.baeldung.kafka.streams; - -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.Serde; -import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.common.utils.Bytes; -import org.apache.kafka.streams.KafkaStreams; -import org.apache.kafka.streams.KeyValue; -import org.apache.kafka.streams.StoreQueryParameters; -import org.apache.kafka.streams.StreamsBuilder; -import org.apache.kafka.streams.StreamsConfig; -import org.apache.kafka.streams.Topology; -import org.apache.kafka.streams.kstream.Consumed; -import org.apache.kafka.streams.kstream.Grouped; -import org.apache.kafka.streams.kstream.JoinWindows; -import org.apache.kafka.streams.kstream.KGroupedStream; -import org.apache.kafka.streams.kstream.KGroupedTable; -import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KTable; -import org.apache.kafka.streams.kstream.Materialized; -import org.apache.kafka.streams.kstream.Produced; -import org.apache.kafka.streams.kstream.TimeWindows; -import org.apache.kafka.streams.state.KeyValueIterator; -import org.apache.kafka.streams.state.KeyValueStore; -import org.apache.kafka.streams.state.QueryableStoreTypes; -import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; -import org.apache.kafka.streams.state.StoreBuilder; -import org.apache.kafka.streams.state.Stores; -import org.apache.kafka.streams.state.WindowStore; -import org.junit.Before; -import org.junit.ClassRule; -import org.junit.Test; -import org.testcontainers.containers.KafkaContainer; -import org.testcontainers.utility.DockerImageName; - -import java.time.Duration; -import java.util.Arrays; -import java.util.Locale; -import java.util.Properties; - -import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG; -import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG; -import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG; - -public class KafkaStreamsLiveTest { - private final String LEFT_TOPIC = "left-stream-topic"; - private final String RIGHT_TOPIC = "right-stream-topic"; - private final String LEFT_RIGHT_TOPIC = "left-right-stream-topic"; - - private KafkaProducer producer = createKafkaProducer(); - private Properties streamsConfiguration = new Properties(); - - static final String TEXT_LINES_TOPIC = "TextLinesTopic"; - - private final String TEXT_EXAMPLE_1 = "test test and test"; - private final String TEXT_EXAMPLE_2 = "test filter filter this sentence"; - - @ClassRule - public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3")); - - @Before - public void setUp() { - streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers()); - streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); - streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); - streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); - streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - } - - @Test - public void shouldTestKafkaTableLatestWord() throws InterruptedException { - String inputTopic = "topicTable"; - - final StreamsBuilder builder = new StreamsBuilder(); - - KTable textLinesTable = builder.table(inputTopic, - Consumed.with(Serdes.String(), Serdes.String())); - - textLinesTable.toStream().foreach((word, count) -> System.out.println("Latest word: " + word + " -> " + count)); - - final Topology topology = builder.build(); - streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "latest-word-id"); - KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); - - streams.cleanUp(); - streams.start(); - producer.send(new ProducerRecord(inputTopic, "1", TEXT_EXAMPLE_1)); - producer.send(new ProducerRecord(inputTopic, "2", TEXT_EXAMPLE_2)); - - Thread.sleep(2000); - streams.close(); - } - - @Test - public void shouldTestWordCountKafkaStreams() throws InterruptedException { - String wordCountTopic = "wordCountTopic"; - - final StreamsBuilder builder = new StreamsBuilder(); - KStream textLines = builder.stream(wordCountTopic, - Consumed.with(Serdes.String(), Serdes.String())); - - KTable wordCounts = textLines - .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT) - .split("\\W+"))) - .groupBy((key, word) -> word) - .count(Materialized.> as("counts-store")); - - wordCounts.toStream().foreach((word, count) -> System.out.println("Word: " + word + " -> " + count)); - - wordCounts.toStream().to("outputTopic", - Produced.with(Serdes.String(), Serdes.Long())); - - streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-stream-table-id"); - final Topology topology = builder.build(); - KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); - - streams.cleanUp(); - streams.start(); - - producer.send(new ProducerRecord(wordCountTopic, "1", TEXT_EXAMPLE_1)); - producer.send(new ProducerRecord(wordCountTopic, "2", TEXT_EXAMPLE_2)); - - Thread.sleep(2000); - streams.close(); - } - - // Filter, map - @Test - public void shouldTestStatelessTransformations() throws InterruptedException { - String wordCountTopic = "wordCountTopic"; - - //when - final StreamsBuilder builder = new StreamsBuilder(); - KStream textLines = builder.stream(wordCountTopic, - Consumed.with(Serdes.String(), Serdes.String())); - - final KStream textLinesUpperCase = - textLines - .map((key, value) -> KeyValue.pair(value, value.toUpperCase())) - .filter((key, value) -> value.contains("FILTER")); - - KTable wordCounts = textLinesUpperCase - .flatMapValues(value -> Arrays.asList(value.split("\\W+"))) - .groupBy((key, word) -> word) - .count(Materialized.> as("counts-store")); - - wordCounts.toStream().foreach((word, count) -> System.out.println("Word: " + word + " -> " + count)); - - streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-filter-map-id"); - final Topology topology = builder.build(); - KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); - - streams.cleanUp(); - streams.start(); - - producer.send(new ProducerRecord(wordCountTopic, "1", TEXT_EXAMPLE_1)); - producer.send(new ProducerRecord(wordCountTopic, "2", TEXT_EXAMPLE_2)); - - Thread.sleep(2000); - streams.close(); - - } - - @Test - public void shouldTestAggregationStatefulTransformations() throws InterruptedException { - String aggregationTopic = "aggregationTopic"; - - final StreamsBuilder builder = new StreamsBuilder(); - final KStream input = builder.stream(aggregationTopic, - Consumed.with(Serdes.ByteArray(), Serdes.String())); - final KTable aggregated = input - .groupBy((key, value) -> (value != null && value.length() > 0) ? value.substring(0, 2).toLowerCase() : "", - Grouped.with(Serdes.String(), Serdes.String())) - .aggregate(() -> 0L, (aggKey, newValue, aggValue) -> aggValue + newValue.length(), - Materialized.with(Serdes.String(), Serdes.Long())); - - aggregated.toStream().foreach((word, count) -> System.out.println("Word: " + word + " -> " + count)); - - streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "aggregation-id"); - final Topology topology = builder.build(); - KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); - - streams.cleanUp(); - streams.start(); - - producer.send(new ProducerRecord(aggregationTopic, "1", "one")); - producer.send(new ProducerRecord(aggregationTopic, "2", "two")); - producer.send(new ProducerRecord(aggregationTopic, "3", "three")); - producer.send(new ProducerRecord(aggregationTopic, "4", "four")); - producer.send(new ProducerRecord(aggregationTopic, "5", "five")); - - Thread.sleep(5000); - streams.close(); - - } - - @Test - public void shouldTestWindowingJoinStatefulTransformations() throws InterruptedException { - final StreamsBuilder builder = new StreamsBuilder(); - - KStream leftSource = builder.stream(LEFT_TOPIC); - KStream rightSource = builder.stream(RIGHT_TOPIC); - - KStream leftRightSource = leftSource.outerJoin(rightSource, - (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, - JoinWindows.of(Duration.ofSeconds(5))) - .groupByKey() - .reduce(((key, lastValue) -> lastValue)) - .toStream(); - - leftRightSource.foreach((key, value) -> System.out.println("(key= " + key + ") -> (" + value + ")")); - - final Topology topology = builder.build(); - streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "windowing-join-id"); - KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); - - streams.cleanUp(); - streams.start(); - - producer.send(new ProducerRecord(LEFT_TOPIC, "1", "left")); - producer.send(new ProducerRecord(RIGHT_TOPIC, "2", "right")); - - Thread.sleep(2000); - streams.close(); - } - - @Test - public void shouldTestWordCountWithInteractiveQueries() throws InterruptedException { - - final Serde stringSerde = Serdes.String(); - final StreamsBuilder builder = new StreamsBuilder(); - final KStream - textLines = builder.stream(TEXT_LINES_TOPIC, Consumed.with(Serdes.String(), Serdes.String())); - - final KGroupedStream groupedByWord = textLines - .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) - .groupBy((key, word) -> word, Grouped.with(stringSerde, stringSerde)); - - groupedByWord.count(Materialized.>as("WordCountsStore") - .withValueSerde(Serdes.Long())); - - streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-interactive-queries"); - - final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration); - streams.cleanUp(); - streams.start(); - - producer.send(new ProducerRecord(TEXT_LINES_TOPIC, "1", TEXT_EXAMPLE_1)); - producer.send(new ProducerRecord(TEXT_LINES_TOPIC, "2", TEXT_EXAMPLE_2)); - - Thread.sleep(2000); - ReadOnlyKeyValueStore keyValueStore = - streams.store(StoreQueryParameters.fromNameAndType( - "WordCountsStore", QueryableStoreTypes.keyValueStore())); - - KeyValueIterator range = keyValueStore.all(); - while (range.hasNext()) { - KeyValue next = range.next(); - System.out.println("Count for " + next.key + ": " + next.value); - } - - streams.close(); - } - - private static KafkaProducer createKafkaProducer() { - - Properties props = new Properties(); - props.put(BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers()); - props.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); - props.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); - - return new KafkaProducer(props); - - } -} - - From 3f01b5fb0a933043aa7128aef1142cc90d2269ce Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:04:17 +0530 Subject: [PATCH 02/12] JAVA-6390: pom and README changes for libraries-data-3 --- .../log4j.properties | 0 libraries-data-3/README.md | 6 +-- libraries-data-3/pom.xml | 43 ------------------- 3 files changed, 2 insertions(+), 47 deletions(-) rename {libraries-data-3 => apache-kafka}/log4j.properties (100%) diff --git a/libraries-data-3/log4j.properties b/apache-kafka/log4j.properties similarity index 100% rename from libraries-data-3/log4j.properties rename to apache-kafka/log4j.properties diff --git a/libraries-data-3/README.md b/libraries-data-3/README.md index d3bfb2c80c..fe856436f1 100644 --- a/libraries-data-3/README.md +++ b/libraries-data-3/README.md @@ -3,9 +3,7 @@ This module contains articles about libraries for data processing in Java. ### 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) -- More articles: [[<-- prev]](/../libraries-data-2) + ##### Building the project -You can build the project from the command line using: *mvn clean install*, or in an IDE. If you have issues with the derive4j imports in your IDE, you have to add the folder: *target/generated-sources/annotations* to the project build path in your IDE. +You can build the project from the command line using: *mvn clean install*, or in an IDE. \ No newline at end of file diff --git a/libraries-data-3/pom.xml b/libraries-data-3/pom.xml index 37d5c7ca0d..1a95613cb3 100644 --- a/libraries-data-3/pom.xml +++ b/libraries-data-3/pom.xml @@ -13,52 +13,9 @@ - - org.apache.kafka - kafka-clients - ${kafka.version} - - - org.apache.kafka - kafka-streams - ${kafka.version} - - - org.slf4j - slf4j-api - ${slf4j.version} - - - org.slf4j - slf4j-log4j12 - ${slf4j.version} - - - org.assertj - assertj-core - ${assertj.version} - test - - - org.testcontainers - kafka - ${testcontainers-kafka.version} - test - - - org.testcontainers - junit-jupiter - ${testcontainers-jupiter.version} - test - - 3.6.2 - 1.7.25 - 2.8.0 - 1.15.3 - 1.15.3 \ No newline at end of file From 0d280b54b977aef16cee5a65cc3f2653e5a96e80 Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:09:04 +0530 Subject: [PATCH 03/12] JAVA-6390: Move kafka articles from libraries-data-3 to new module apache-kafka --- .../com/baeldung/flink/FlinkDataPipeline.java | 70 ++++++++++++++++ .../baeldung/flink/connector/Consumers.java | 23 +++--- .../baeldung/flink/connector/Producers.java | 0 .../java/com/baeldung/flink/model/Backup.java | 0 .../baeldung/flink/model/InputMessage.java | 12 +-- .../flink/operator/BackupAggregator.java | 34 ++++++++ .../InputMessageTimestampAssigner.java | 4 +- .../flink/operator/WordsCapitalizer.java | 0 .../schema/BackupSerializationSchema.java | 5 +- .../InputMessageDeserializationSchema.java | 4 +- .../kafka/consumer/CountryPopulation.java | 0 .../consumer/CountryPopulationConsumer.java | 4 +- .../kafka/producer/EvenOddPartitioner.java | 0 .../kafka/producer/KafkaProducer.java | 4 +- .../flink/BackupCreatorIntegrationTest.java | 0 .../flink/WordCapitalizerIntegrationTest.java | 0 .../CountryPopulationConsumerUnitTest.java | 0 .../kafka/producer/KafkaProducerUnitTest.java | 0 .../com/baeldung/flink/FlinkDataPipeline.java | 82 ------------------- .../flink/operator/BackupAggregator.java | 34 -------- 20 files changed, 128 insertions(+), 148 deletions(-) create mode 100644 apache-kafka/src/main/java/com/baeldung/flink/FlinkDataPipeline.java rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/connector/Consumers.java (53%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/connector/Producers.java (100%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/model/Backup.java (100%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/model/InputMessage.java (80%) create mode 100644 apache-kafka/src/main/java/com/baeldung/flink/operator/BackupAggregator.java rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java (88%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/operator/WordsCapitalizer.java (100%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java (90%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java (89%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/kafka/consumer/CountryPopulation.java (100%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java (89%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/kafka/producer/EvenOddPartitioner.java (100%) rename {libraries-data-2 => apache-kafka}/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java (95%) rename {libraries-data-2 => apache-kafka}/src/test/java/com/baeldung/flink/BackupCreatorIntegrationTest.java (100%) rename {libraries-data-2 => apache-kafka}/src/test/java/com/baeldung/flink/WordCapitalizerIntegrationTest.java (100%) rename {libraries-data-2 => apache-kafka}/src/test/java/com/baeldung/kafka/consumer/CountryPopulationConsumerUnitTest.java (100%) rename {libraries-data-2 => apache-kafka}/src/test/java/com/baeldung/kafka/producer/KafkaProducerUnitTest.java (100%) delete mode 100644 libraries-data-2/src/main/java/com/baeldung/flink/FlinkDataPipeline.java delete mode 100644 libraries-data-2/src/main/java/com/baeldung/flink/operator/BackupAggregator.java diff --git a/apache-kafka/src/main/java/com/baeldung/flink/FlinkDataPipeline.java b/apache-kafka/src/main/java/com/baeldung/flink/FlinkDataPipeline.java new file mode 100644 index 0000000000..4502b628b2 --- /dev/null +++ b/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 flinkKafkaConsumer = createStringConsumerForTopic(inputTopic, address, consumerGroup); + flinkKafkaConsumer.setStartFromEarliest(); + + DataStream stringInputStream = environment.addSource(flinkKafkaConsumer); + + FlinkKafkaProducer011 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 flinkKafkaConsumer = createInputMessageConsumer(inputTopic, kafkaAddress, consumerGroup); + flinkKafkaConsumer.setStartFromEarliest(); + + flinkKafkaConsumer.assignTimestampsAndWatermarks(new InputMessageTimestampAssigner()); + FlinkKafkaProducer011 flinkKafkaProducer = createBackupProducer(outputTopic, kafkaAddress); + + DataStream 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(); + } + +} diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/connector/Consumers.java b/apache-kafka/src/main/java/com/baeldung/flink/connector/Consumers.java similarity index 53% rename from libraries-data-2/src/main/java/com/baeldung/flink/connector/Consumers.java rename to apache-kafka/src/main/java/com/baeldung/flink/connector/Consumers.java index 514085f9c4..c72cb8a2d6 100644 --- a/libraries-data-2/src/main/java/com/baeldung/flink/connector/Consumers.java +++ b/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 createStringConsumerForTopic( - String topic, String kafkaAddress, String kafkaGroup ) { - Properties props = new Properties(); - props.setProperty("bootstrap.servers", kafkaAddress); - props.setProperty("group.id",kafkaGroup); - FlinkKafkaConsumer011 consumer = - new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(),props); + public static FlinkKafkaConsumer011 createStringConsumerForTopic(String topic, String kafkaAddress, String kafkaGroup) { + Properties props = new Properties(); + props.setProperty("bootstrap.servers", kafkaAddress); + props.setProperty("group.id", kafkaGroup); + FlinkKafkaConsumer011 consumer = new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(), props); - return consumer; -} + return consumer; + } - public static FlinkKafkaConsumer011 createInputMessageConsumer(String topic, String kafkaAddress, String kafkaGroup ) { + public static FlinkKafkaConsumer011 createInputMessageConsumer(String topic, String kafkaAddress, String kafkaGroup) { Properties properties = new Properties(); properties.setProperty("bootstrap.servers", kafkaAddress); - properties.setProperty("group.id",kafkaGroup); - FlinkKafkaConsumer011 consumer = new FlinkKafkaConsumer011( - topic, new InputMessageDeserializationSchema(),properties); + properties.setProperty("group.id", kafkaGroup); + FlinkKafkaConsumer011 consumer = new FlinkKafkaConsumer011(topic, new InputMessageDeserializationSchema(), properties); return consumer; } diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/connector/Producers.java b/apache-kafka/src/main/java/com/baeldung/flink/connector/Producers.java similarity index 100% rename from libraries-data-2/src/main/java/com/baeldung/flink/connector/Producers.java rename to apache-kafka/src/main/java/com/baeldung/flink/connector/Producers.java diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/model/Backup.java b/apache-kafka/src/main/java/com/baeldung/flink/model/Backup.java similarity index 100% rename from libraries-data-2/src/main/java/com/baeldung/flink/model/Backup.java rename to apache-kafka/src/main/java/com/baeldung/flink/model/Backup.java diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/model/InputMessage.java b/apache-kafka/src/main/java/com/baeldung/flink/model/InputMessage.java similarity index 80% rename from libraries-data-2/src/main/java/com/baeldung/flink/model/InputMessage.java rename to apache-kafka/src/main/java/com/baeldung/flink/model/InputMessage.java index b3f75256ae..9331811b91 100644 --- a/libraries-data-2/src/main/java/com/baeldung/flink/model/InputMessage.java +++ b/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,13 +56,12 @@ 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) && - Objects.equal(message, message1.message); + return Objects.equal(sender, message1.sender) && Objects.equal(recipient, message1.recipient) && Objects.equal(sentAt, message1.sentAt) && Objects.equal(message, message1.message); } @Override diff --git a/apache-kafka/src/main/java/com/baeldung/flink/operator/BackupAggregator.java b/apache-kafka/src/main/java/com/baeldung/flink/operator/BackupAggregator.java new file mode 100644 index 0000000000..bac1c8c705 --- /dev/null +++ b/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, Backup> { + @Override + public List createAccumulator() { + return new ArrayList<>(); + } + + @Override + public List add(InputMessage inputMessage, List inputMessages) { + inputMessages.add(inputMessage); + return inputMessages; + } + + @Override + public Backup getResult(List inputMessages) { + Backup backup = new Backup(inputMessages, LocalDateTime.now()); + return backup; + } + + @Override + public List merge(List inputMessages, List acc1) { + inputMessages.addAll(acc1); + return inputMessages; + } +} diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java b/apache-kafka/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java similarity index 88% rename from libraries-data-2/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java rename to apache-kafka/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java index 05828d9588..995fe41717 100644 --- a/libraries-data-2/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java +++ b/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 diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/operator/WordsCapitalizer.java b/apache-kafka/src/main/java/com/baeldung/flink/operator/WordsCapitalizer.java similarity index 100% rename from libraries-data-2/src/main/java/com/baeldung/flink/operator/WordsCapitalizer.java rename to apache-kafka/src/main/java/com/baeldung/flink/operator/WordsCapitalizer.java diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java b/apache-kafka/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java similarity index 90% rename from libraries-data-2/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java rename to apache-kafka/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java index 967b266bb6..d4b7b0955a 100644 --- a/libraries-data-2/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java +++ b/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 { +public class BackupSerializationSchema implements SerializationSchema { 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()); } diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java b/apache-kafka/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java similarity index 89% rename from libraries-data-2/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java rename to apache-kafka/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java index 9aaf8b9877..e521af7c2d 100644 --- a/libraries-data-2/src/main/java/com/baeldung/flink/schema/InputMessageDeserializationSchema.java +++ b/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 { +public class InputMessageDeserializationSchema implements DeserializationSchema { static ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule()); - @Override public InputMessage deserialize(byte[] bytes) throws IOException { diff --git a/libraries-data-2/src/main/java/com/baeldung/kafka/consumer/CountryPopulation.java b/apache-kafka/src/main/java/com/baeldung/kafka/consumer/CountryPopulation.java similarity index 100% rename from libraries-data-2/src/main/java/com/baeldung/kafka/consumer/CountryPopulation.java rename to apache-kafka/src/main/java/com/baeldung/kafka/consumer/CountryPopulation.java diff --git a/libraries-data-2/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java b/apache-kafka/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java similarity index 89% rename from libraries-data-2/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java rename to apache-kafka/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java index ba4dfe6f3b..a67d3a581c 100644 --- a/libraries-data-2/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java +++ b/apache-kafka/src/main/java/com/baeldung/kafka/consumer/CountryPopulationConsumer.java @@ -19,9 +19,7 @@ public class CountryPopulationConsumer { private java.util.function.Consumer exceptionConsumer; private java.util.function.Consumer countryPopulationConsumer; - public CountryPopulationConsumer( - Consumer consumer, java.util.function.Consumer exceptionConsumer, - java.util.function.Consumer countryPopulationConsumer) { + public CountryPopulationConsumer(Consumer consumer, java.util.function.Consumer exceptionConsumer, java.util.function.Consumer countryPopulationConsumer) { this.consumer = consumer; this.exceptionConsumer = exceptionConsumer; this.countryPopulationConsumer = countryPopulationConsumer; diff --git a/libraries-data-2/src/main/java/com/baeldung/kafka/producer/EvenOddPartitioner.java b/apache-kafka/src/main/java/com/baeldung/kafka/producer/EvenOddPartitioner.java similarity index 100% rename from libraries-data-2/src/main/java/com/baeldung/kafka/producer/EvenOddPartitioner.java rename to apache-kafka/src/main/java/com/baeldung/kafka/producer/EvenOddPartitioner.java diff --git a/libraries-data-2/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java b/apache-kafka/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java similarity index 95% rename from libraries-data-2/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java rename to apache-kafka/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java index 911c9ed3d7..fa508593e0 100644 --- a/libraries-data-2/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java +++ b/apache-kafka/src/main/java/com/baeldung/kafka/producer/KafkaProducer.java @@ -15,8 +15,7 @@ public class KafkaProducer { } public Future 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(); } - } diff --git a/libraries-data-2/src/test/java/com/baeldung/flink/BackupCreatorIntegrationTest.java b/apache-kafka/src/test/java/com/baeldung/flink/BackupCreatorIntegrationTest.java similarity index 100% rename from libraries-data-2/src/test/java/com/baeldung/flink/BackupCreatorIntegrationTest.java rename to apache-kafka/src/test/java/com/baeldung/flink/BackupCreatorIntegrationTest.java diff --git a/libraries-data-2/src/test/java/com/baeldung/flink/WordCapitalizerIntegrationTest.java b/apache-kafka/src/test/java/com/baeldung/flink/WordCapitalizerIntegrationTest.java similarity index 100% rename from libraries-data-2/src/test/java/com/baeldung/flink/WordCapitalizerIntegrationTest.java rename to apache-kafka/src/test/java/com/baeldung/flink/WordCapitalizerIntegrationTest.java diff --git a/libraries-data-2/src/test/java/com/baeldung/kafka/consumer/CountryPopulationConsumerUnitTest.java b/apache-kafka/src/test/java/com/baeldung/kafka/consumer/CountryPopulationConsumerUnitTest.java similarity index 100% rename from libraries-data-2/src/test/java/com/baeldung/kafka/consumer/CountryPopulationConsumerUnitTest.java rename to apache-kafka/src/test/java/com/baeldung/kafka/consumer/CountryPopulationConsumerUnitTest.java diff --git a/libraries-data-2/src/test/java/com/baeldung/kafka/producer/KafkaProducerUnitTest.java b/apache-kafka/src/test/java/com/baeldung/kafka/producer/KafkaProducerUnitTest.java similarity index 100% rename from libraries-data-2/src/test/java/com/baeldung/kafka/producer/KafkaProducerUnitTest.java rename to apache-kafka/src/test/java/com/baeldung/kafka/producer/KafkaProducerUnitTest.java diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/FlinkDataPipeline.java b/libraries-data-2/src/main/java/com/baeldung/flink/FlinkDataPipeline.java deleted file mode 100644 index d02b1bcb83..0000000000 --- a/libraries-data-2/src/main/java/com/baeldung/flink/FlinkDataPipeline.java +++ /dev/null @@ -1,82 +0,0 @@ -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 flinkKafkaConsumer = - createStringConsumerForTopic(inputTopic, address, consumerGroup); - flinkKafkaConsumer.setStartFromEarliest(); - - DataStream stringInputStream = - environment.addSource(flinkKafkaConsumer); - - FlinkKafkaProducer011 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 flinkKafkaConsumer = - createInputMessageConsumer(inputTopic, kafkaAddress, consumerGroup); - flinkKafkaConsumer.setStartFromEarliest(); - - flinkKafkaConsumer - .assignTimestampsAndWatermarks(new InputMessageTimestampAssigner()); - FlinkKafkaProducer011 flinkKafkaProducer = - createBackupProducer(outputTopic, kafkaAddress); - - DataStream 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(); - } - -} diff --git a/libraries-data-2/src/main/java/com/baeldung/flink/operator/BackupAggregator.java b/libraries-data-2/src/main/java/com/baeldung/flink/operator/BackupAggregator.java deleted file mode 100644 index c39b8413d1..0000000000 --- a/libraries-data-2/src/main/java/com/baeldung/flink/operator/BackupAggregator.java +++ /dev/null @@ -1,34 +0,0 @@ -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, Backup> { - @Override - public List createAccumulator() { - return new ArrayList<>(); - } - - @Override - public List add(InputMessage inputMessage, List inputMessages) { - inputMessages.add(inputMessage); - return inputMessages; - } - - @Override - public Backup getResult(List inputMessages) { - Backup backup = new Backup(inputMessages, LocalDateTime.now()); - return backup; - } - - @Override - public List merge(List inputMessages, List acc1) { - inputMessages.addAll(acc1); - return inputMessages; - } - } From 2caf4b52f38d1faaba19c7c4b1df5c640cd8fa8d Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:09:48 +0530 Subject: [PATCH 04/12] JAVA-6390: pom and README changes for libraries-data-2 --- libraries-data-2/README.md | 2 -- libraries-data-2/pom.xml | 13 ------------- 2 files changed, 15 deletions(-) diff --git a/libraries-data-2/README.md b/libraries-data-2/README.md index 893d3e64e8..ee604acf6b 100644 --- a/libraries-data-2/README.md +++ b/libraries-data-2/README.md @@ -11,8 +11,6 @@ This module contains articles about libraries for data processing in Java. - [An Introduction to SuanShu](https://www.baeldung.com/suanshu) - [Intro to Derive4J](https://www.baeldung.com/derive4j) - [Univocity Parsers](https://www.baeldung.com/java-univocity-parsers) -- [Using Kafka MockConsumer](https://www.baeldung.com/kafka-mockconsumer) -- [Using Kafka MockProducer](https://www.baeldung.com/kafka-mockproducer) - More articles: [[<-- prev]](/../libraries-data) ##### Building the project diff --git a/libraries-data-2/pom.xml b/libraries-data-2/pom.xml index 1feb3142ee..cce2e57d22 100644 --- a/libraries-data-2/pom.xml +++ b/libraries-data-2/pom.xml @@ -116,11 +116,6 @@ univocity-parsers ${univocity.version} - - org.apache.kafka - kafka-clients - ${kafka.version} - com.google.guava guava @@ -144,13 +139,6 @@ ${byte-buddy.version} test - - org.apache.kafka - kafka-clients - ${kafka.version} - test - test - @@ -176,7 +164,6 @@ 1.7.25 3.0.0 2.8.4 - 2.5.0 29.0-jre From c0cae3571460000b2d9a3b4730415ce37f8583b8 Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:15:36 +0530 Subject: [PATCH 05/12] JAVA-6390: Move kafka articles from libraries-data to new module apache-kafka --- .../connect-file-sink.properties | 0 .../connect-file-source.properties | 0 .../connect-standalone.properties | 0 .../connect-distributed.properties | 0 .../02_Distributed/connect-file-sink.json | 0 .../02_Distributed/connect-file-source.json | 0 .../connect-distributed.properties | 0 .../connect-file-source-transform.json | 0 .../04_Custom/connect-mongodb-sink.json | 0 .../04_Custom/connect-mqtt-source.json | 0 .../04_Custom/docker-compose.yaml | 0 .../KafkaStreamsLiveTest.java | 279 ++++++++++++++++++ .../kafkastreams/KafkaStreamsLiveTest.java | 62 ---- 13 files changed, 279 insertions(+), 62 deletions(-) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/01_Quick_Start/connect-file-sink.properties (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/01_Quick_Start/connect-file-source.properties (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/01_Quick_Start/connect-standalone.properties (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/02_Distributed/connect-distributed.properties (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/02_Distributed/connect-file-sink.json (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/02_Distributed/connect-file-source.json (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/03_Transform/connect-distributed.properties (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/03_Transform/connect-file-source-transform.json (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/04_Custom/connect-mongodb-sink.json (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/04_Custom/connect-mqtt-source.json (100%) rename {libraries-data => apache-kafka}/src/main/resources/kafka-connect/04_Custom/docker-compose.yaml (100%) create mode 100644 apache-kafka/src/test/java/com/baeldung/kafka/streamsvsconsumer/KafkaStreamsLiveTest.java delete mode 100644 libraries-data/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java diff --git a/libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-file-sink.properties b/apache-kafka/src/main/resources/kafka-connect/01_Quick_Start/connect-file-sink.properties similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-file-sink.properties rename to apache-kafka/src/main/resources/kafka-connect/01_Quick_Start/connect-file-sink.properties diff --git a/libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-file-source.properties b/apache-kafka/src/main/resources/kafka-connect/01_Quick_Start/connect-file-source.properties similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-file-source.properties rename to apache-kafka/src/main/resources/kafka-connect/01_Quick_Start/connect-file-source.properties diff --git a/libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-standalone.properties b/apache-kafka/src/main/resources/kafka-connect/01_Quick_Start/connect-standalone.properties similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-standalone.properties rename to apache-kafka/src/main/resources/kafka-connect/01_Quick_Start/connect-standalone.properties diff --git a/libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-distributed.properties b/apache-kafka/src/main/resources/kafka-connect/02_Distributed/connect-distributed.properties similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-distributed.properties rename to apache-kafka/src/main/resources/kafka-connect/02_Distributed/connect-distributed.properties diff --git a/libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-file-sink.json b/apache-kafka/src/main/resources/kafka-connect/02_Distributed/connect-file-sink.json similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-file-sink.json rename to apache-kafka/src/main/resources/kafka-connect/02_Distributed/connect-file-sink.json diff --git a/libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-file-source.json b/apache-kafka/src/main/resources/kafka-connect/02_Distributed/connect-file-source.json similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-file-source.json rename to apache-kafka/src/main/resources/kafka-connect/02_Distributed/connect-file-source.json diff --git a/libraries-data/src/main/resources/kafka-connect/03_Transform/connect-distributed.properties b/apache-kafka/src/main/resources/kafka-connect/03_Transform/connect-distributed.properties similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/03_Transform/connect-distributed.properties rename to apache-kafka/src/main/resources/kafka-connect/03_Transform/connect-distributed.properties diff --git a/libraries-data/src/main/resources/kafka-connect/03_Transform/connect-file-source-transform.json b/apache-kafka/src/main/resources/kafka-connect/03_Transform/connect-file-source-transform.json similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/03_Transform/connect-file-source-transform.json rename to apache-kafka/src/main/resources/kafka-connect/03_Transform/connect-file-source-transform.json diff --git a/libraries-data/src/main/resources/kafka-connect/04_Custom/connect-mongodb-sink.json b/apache-kafka/src/main/resources/kafka-connect/04_Custom/connect-mongodb-sink.json similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/04_Custom/connect-mongodb-sink.json rename to apache-kafka/src/main/resources/kafka-connect/04_Custom/connect-mongodb-sink.json diff --git a/libraries-data/src/main/resources/kafka-connect/04_Custom/connect-mqtt-source.json b/apache-kafka/src/main/resources/kafka-connect/04_Custom/connect-mqtt-source.json similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/04_Custom/connect-mqtt-source.json rename to apache-kafka/src/main/resources/kafka-connect/04_Custom/connect-mqtt-source.json diff --git a/libraries-data/src/main/resources/kafka-connect/04_Custom/docker-compose.yaml b/apache-kafka/src/main/resources/kafka-connect/04_Custom/docker-compose.yaml similarity index 100% rename from libraries-data/src/main/resources/kafka-connect/04_Custom/docker-compose.yaml rename to apache-kafka/src/main/resources/kafka-connect/04_Custom/docker-compose.yaml diff --git a/apache-kafka/src/test/java/com/baeldung/kafka/streamsvsconsumer/KafkaStreamsLiveTest.java b/apache-kafka/src/test/java/com/baeldung/kafka/streamsvsconsumer/KafkaStreamsLiveTest.java new file mode 100644 index 0000000000..88de6101dc --- /dev/null +++ b/apache-kafka/src/test/java/com/baeldung/kafka/streamsvsconsumer/KafkaStreamsLiveTest.java @@ -0,0 +1,279 @@ +package com.baeldung.kafka.streamsvsconsumer; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StoreQueryParameters; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.kstream.Consumed; +import org.apache.kafka.streams.kstream.Grouped; +import org.apache.kafka.streams.kstream.JoinWindows; +import org.apache.kafka.streams.kstream.KGroupedStream; +import org.apache.kafka.streams.kstream.KGroupedTable; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.Produced; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; +import org.apache.kafka.streams.state.QueryableStoreTypes; +import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; +import org.apache.kafka.streams.state.StoreBuilder; +import org.apache.kafka.streams.state.Stores; +import org.apache.kafka.streams.state.WindowStore; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import org.testcontainers.containers.KafkaContainer; +import org.testcontainers.utility.DockerImageName; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Locale; +import java.util.Properties; + +import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG; + +public class KafkaStreamsLiveTest { + private final String LEFT_TOPIC = "left-stream-topic"; + private final String RIGHT_TOPIC = "right-stream-topic"; + private final String LEFT_RIGHT_TOPIC = "left-right-stream-topic"; + + private KafkaProducer producer = createKafkaProducer(); + private Properties streamsConfiguration = new Properties(); + + static final String TEXT_LINES_TOPIC = "TextLinesTopic"; + + private final String TEXT_EXAMPLE_1 = "test test and test"; + private final String TEXT_EXAMPLE_2 = "test filter filter this sentence"; + + @ClassRule + public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3")); + + @Before + public void setUp() { + streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers()); + streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); + streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); + streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); + streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + } + + @Test + public void shouldTestKafkaTableLatestWord() throws InterruptedException { + String inputTopic = "topicTable"; + + final StreamsBuilder builder = new StreamsBuilder(); + + KTable textLinesTable = builder.table(inputTopic, + Consumed.with(Serdes.String(), Serdes.String())); + + textLinesTable.toStream().foreach((word, count) -> System.out.println("Latest word: " + word + " -> " + count)); + + final Topology topology = builder.build(); + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "latest-word-id"); + KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); + + streams.cleanUp(); + streams.start(); + producer.send(new ProducerRecord(inputTopic, "1", TEXT_EXAMPLE_1)); + producer.send(new ProducerRecord(inputTopic, "2", TEXT_EXAMPLE_2)); + + Thread.sleep(2000); + streams.close(); + } + + @Test + public void shouldTestWordCountKafkaStreams() throws InterruptedException { + String wordCountTopic = "wordCountTopic"; + + final StreamsBuilder builder = new StreamsBuilder(); + KStream textLines = builder.stream(wordCountTopic, + Consumed.with(Serdes.String(), Serdes.String())); + + KTable wordCounts = textLines + .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT) + .split("\\W+"))) + .groupBy((key, word) -> word) + .count(Materialized.> as("counts-store")); + + wordCounts.toStream().foreach((word, count) -> System.out.println("Word: " + word + " -> " + count)); + + wordCounts.toStream().to("outputTopic", + Produced.with(Serdes.String(), Serdes.Long())); + + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-stream-table-id"); + final Topology topology = builder.build(); + KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); + + streams.cleanUp(); + streams.start(); + + producer.send(new ProducerRecord(wordCountTopic, "1", TEXT_EXAMPLE_1)); + producer.send(new ProducerRecord(wordCountTopic, "2", TEXT_EXAMPLE_2)); + + Thread.sleep(2000); + streams.close(); + } + + // Filter, map + @Test + public void shouldTestStatelessTransformations() throws InterruptedException { + String wordCountTopic = "wordCountTopic"; + + //when + final StreamsBuilder builder = new StreamsBuilder(); + KStream textLines = builder.stream(wordCountTopic, + Consumed.with(Serdes.String(), Serdes.String())); + + final KStream textLinesUpperCase = + textLines + .map((key, value) -> KeyValue.pair(value, value.toUpperCase())) + .filter((key, value) -> value.contains("FILTER")); + + KTable wordCounts = textLinesUpperCase + .flatMapValues(value -> Arrays.asList(value.split("\\W+"))) + .groupBy((key, word) -> word) + .count(Materialized.> as("counts-store")); + + wordCounts.toStream().foreach((word, count) -> System.out.println("Word: " + word + " -> " + count)); + + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-filter-map-id"); + final Topology topology = builder.build(); + KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); + + streams.cleanUp(); + streams.start(); + + producer.send(new ProducerRecord(wordCountTopic, "1", TEXT_EXAMPLE_1)); + producer.send(new ProducerRecord(wordCountTopic, "2", TEXT_EXAMPLE_2)); + + Thread.sleep(2000); + streams.close(); + + } + + @Test + public void shouldTestAggregationStatefulTransformations() throws InterruptedException { + String aggregationTopic = "aggregationTopic"; + + final StreamsBuilder builder = new StreamsBuilder(); + final KStream input = builder.stream(aggregationTopic, + Consumed.with(Serdes.ByteArray(), Serdes.String())); + final KTable aggregated = input + .groupBy((key, value) -> (value != null && value.length() > 0) ? value.substring(0, 2).toLowerCase() : "", + Grouped.with(Serdes.String(), Serdes.String())) + .aggregate(() -> 0L, (aggKey, newValue, aggValue) -> aggValue + newValue.length(), + Materialized.with(Serdes.String(), Serdes.Long())); + + aggregated.toStream().foreach((word, count) -> System.out.println("Word: " + word + " -> " + count)); + + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "aggregation-id"); + final Topology topology = builder.build(); + KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); + + streams.cleanUp(); + streams.start(); + + producer.send(new ProducerRecord(aggregationTopic, "1", "one")); + producer.send(new ProducerRecord(aggregationTopic, "2", "two")); + producer.send(new ProducerRecord(aggregationTopic, "3", "three")); + producer.send(new ProducerRecord(aggregationTopic, "4", "four")); + producer.send(new ProducerRecord(aggregationTopic, "5", "five")); + + Thread.sleep(5000); + streams.close(); + + } + + @Test + public void shouldTestWindowingJoinStatefulTransformations() throws InterruptedException { + final StreamsBuilder builder = new StreamsBuilder(); + + KStream leftSource = builder.stream(LEFT_TOPIC); + KStream rightSource = builder.stream(RIGHT_TOPIC); + + KStream leftRightSource = leftSource.outerJoin(rightSource, + (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, + JoinWindows.of(Duration.ofSeconds(5))) + .groupByKey() + .reduce(((key, lastValue) -> lastValue)) + .toStream(); + + leftRightSource.foreach((key, value) -> System.out.println("(key= " + key + ") -> (" + value + ")")); + + final Topology topology = builder.build(); + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "windowing-join-id"); + KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); + + streams.cleanUp(); + streams.start(); + + producer.send(new ProducerRecord(LEFT_TOPIC, "1", "left")); + producer.send(new ProducerRecord(RIGHT_TOPIC, "2", "right")); + + Thread.sleep(2000); + streams.close(); + } + + @Test + public void shouldTestWordCountWithInteractiveQueries() throws InterruptedException { + + final Serde stringSerde = Serdes.String(); + final StreamsBuilder builder = new StreamsBuilder(); + final KStream + textLines = builder.stream(TEXT_LINES_TOPIC, Consumed.with(Serdes.String(), Serdes.String())); + + final KGroupedStream groupedByWord = textLines + .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .groupBy((key, word) -> word, Grouped.with(stringSerde, stringSerde)); + + groupedByWord.count(Materialized.>as("WordCountsStore") + .withValueSerde(Serdes.Long())); + + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-interactive-queries"); + + final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration); + streams.cleanUp(); + streams.start(); + + producer.send(new ProducerRecord(TEXT_LINES_TOPIC, "1", TEXT_EXAMPLE_1)); + producer.send(new ProducerRecord(TEXT_LINES_TOPIC, "2", TEXT_EXAMPLE_2)); + + Thread.sleep(2000); + ReadOnlyKeyValueStore keyValueStore = + streams.store(StoreQueryParameters.fromNameAndType( + "WordCountsStore", QueryableStoreTypes.keyValueStore())); + + KeyValueIterator range = keyValueStore.all(); + while (range.hasNext()) { + KeyValue next = range.next(); + System.out.println("Count for " + next.key + ": " + next.value); + } + + streams.close(); + } + + private static KafkaProducer createKafkaProducer() { + + Properties props = new Properties(); + props.put(BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers()); + props.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); + props.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); + + return new KafkaProducer(props); + + } +} + + diff --git a/libraries-data/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java b/libraries-data/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java deleted file mode 100644 index 32568e9ea5..0000000000 --- a/libraries-data/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java +++ /dev/null @@ -1,62 +0,0 @@ -package com.baeldung.kafkastreams; - -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.common.serialization.Serde; -import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.streams.KafkaStreams; -import org.apache.kafka.streams.StreamsConfig; -import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KStreamBuilder; -import org.apache.kafka.streams.kstream.KTable; -import org.apache.kafka.test.TestUtils; -import org.junit.Ignore; -import org.junit.Test; - -import java.util.Arrays; -import java.util.Properties; -import java.util.regex.Pattern; - -public class KafkaStreamsLiveTest { - private String bootstrapServers = "localhost:9092"; - - @Test - @Ignore("it needs to have kafka broker running on local") - public void shouldTestKafkaStreams() throws InterruptedException { - //given - String inputTopic = "inputTopic"; - - Properties streamsConfiguration = new Properties(); - streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-live-test"); - streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); - streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); - streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); - streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); - streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - // Use a temporary directory for storing state, which will be automatically removed after the test. - streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath()); - - //when - KStreamBuilder builder = new KStreamBuilder(); - KStream textLines = builder.stream(inputTopic); - Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS); - - KTable wordCounts = textLines - .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase()))) - .groupBy((key, word) -> word) - .count(); - - wordCounts.foreach((word, count) -> System.out.println("word: " + word + " -> " + count)); - - String outputTopic = "outputTopic"; - final Serde stringSerde = Serdes.String(); - final Serde longSerde = Serdes.Long(); - wordCounts.to(stringSerde, longSerde, outputTopic); - - KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration); - streams.start(); - - //then - Thread.sleep(30000); - streams.close(); - } -} From d13f26e437501e2ed80e9ea589c919d1cc4ad6c5 Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:16:20 +0530 Subject: [PATCH 06/12] JAVA-6390: README changes for libraries-data --- libraries-data/README.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/libraries-data/README.md b/libraries-data/README.md index 44fddfd90e..7e87475328 100644 --- a/libraries-data/README.md +++ b/libraries-data/README.md @@ -3,14 +3,10 @@ This module contains articles about libraries for data processing in Java. ### Relevant articles -- [Introduction to KafkaStreams in Java](https://www.baeldung.com/java-kafka-streams) - [Introduction to JCache](https://www.baeldung.com/jcache) - [A Guide to Apache Ignite](https://www.baeldung.com/apache-ignite) - [Apache Ignite with Spring Data](https://www.baeldung.com/apache-ignite-spring-data) - [A Guide to Apache Crunch](https://www.baeldung.com/apache-crunch) - [Intro to Apache Storm](https://www.baeldung.com/apache-storm) -- [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) - [Guide to JMapper](https://www.baeldung.com/jmapper) More articles: [[next -->]](/../libraries-data-2) \ No newline at end of file From b4b48d4268313c01406955a32263cca926f88778 Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:28:59 +0530 Subject: [PATCH 07/12] JAVA-6390: Move kafka articles from libraries-6 to new module apache-kafka --- .../TransactionalMessageProducer.java | 10 +++++----- .../exactlyonce}/TransactionalWordCount.java | 15 ++++++++------- .../com/baeldung/kafka/exactlyonce}/Tuple.java | 6 +++--- 3 files changed, 16 insertions(+), 15 deletions(-) rename {libraries-6/src/main/java/com/baeldung/kafka => apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce}/TransactionalMessageProducer.java (87%) rename {libraries-6/src/main/java/com/baeldung/kafka => apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce}/TransactionalWordCount.java (90%) rename {libraries-6/src/main/java/com/baeldung/kafka => apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce}/Tuple.java (69%) diff --git a/libraries-6/src/main/java/com/baeldung/kafka/TransactionalMessageProducer.java b/apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/TransactionalMessageProducer.java similarity index 87% rename from libraries-6/src/main/java/com/baeldung/kafka/TransactionalMessageProducer.java rename to apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/TransactionalMessageProducer.java index 15488bbaf4..8f2fe6e309 100644 --- a/libraries-6/src/main/java/com/baeldung/kafka/TransactionalMessageProducer.java +++ b/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("input", null, s))); + Stream.of(DATA_MESSAGE_1, DATA_MESSAGE_2) + .forEach(s -> producer.send(new ProducerRecord("input", null, s))); producer.commitTransaction(); - }catch (KafkaException e){ + } catch (KafkaException e) { producer.abortTransaction(); diff --git a/libraries-6/src/main/java/com/baeldung/kafka/TransactionalWordCount.java b/apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/TransactionalWordCount.java similarity index 90% rename from libraries-6/src/main/java/com/baeldung/kafka/TransactionalWordCount.java rename to apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/TransactionalWordCount.java index 0563ba6684..b9a2cb9f85 100644 --- a/libraries-6/src/main/java/com/baeldung/kafka/TransactionalWordCount.java +++ b/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 records = consumer.poll(ofSeconds(60)); Map 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> 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 createKafkaConsumer() { diff --git a/libraries-6/src/main/java/com/baeldung/kafka/Tuple.java b/apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/Tuple.java similarity index 69% rename from libraries-6/src/main/java/com/baeldung/kafka/Tuple.java rename to apache-kafka/src/main/java/com/baeldung/kafka/exactlyonce/Tuple.java index 883de4ba21..ad61e905fd 100644 --- a/libraries-6/src/main/java/com/baeldung/kafka/Tuple.java +++ b/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() { From 9b2813f0f7925354c5e57ce7035c38bb6605ea9a Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:30:14 +0530 Subject: [PATCH 08/12] JAVA-6390: pom and README changes for libraries-6 --- libraries-6/README.md | 1 - libraries-6/pom.xml | 13 ------------- 2 files changed, 14 deletions(-) diff --git a/libraries-6/README.md b/libraries-6/README.md index ecad499e07..210a76ca87 100644 --- a/libraries-6/README.md +++ b/libraries-6/README.md @@ -13,7 +13,6 @@ Remember, for advanced libraries like [Jackson](/jackson) and [JUnit](/testing-m - [Implementing a FTP-Client in Java](https://www.baeldung.com/java-ftp-client) - [Introduction to Functional Java](https://www.baeldung.com/java-functional-library) - [A Guide to the Reflections Library](https://www.baeldung.com/reflections-library) -- [Exactly Once Processing in Kafka with Java](https://www.baeldung.com/kafka-exactly-once) - [Introduction to Protonpack](https://www.baeldung.com/java-protonpack) - [Java-R Integration](https://www.baeldung.com/java-r-integration) - [Using libphonenumber to Validate Phone Numbers](https://www.baeldung.com/java-libphonenumber) diff --git a/libraries-6/pom.xml b/libraries-6/pom.xml index 8d05d2013c..d2ed06af6a 100644 --- a/libraries-6/pom.xml +++ b/libraries-6/pom.xml @@ -22,18 +22,6 @@ protonpack ${protonpack.version} - - org.apache.kafka - kafka-streams - ${kafka.version} - - - org.apache.kafka - kafka-clients - ${kafka.version} - test - test - io.github.resilience4j resilience4j-circuitbreaker @@ -148,7 +136,6 @@ - 2.0.0 1.10.0 0.9.11 2.7.1 From 5eec2c4bbaf85ac3d022192794c0479c481711ba Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:31:33 +0530 Subject: [PATCH 09/12] JAVA-6390: pom and README for apache-kafka --- apache-kafka/README.md | 18 ++++ apache-kafka/pom.xml | 181 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 199 insertions(+) create mode 100644 apache-kafka/README.md create mode 100644 apache-kafka/pom.xml diff --git a/apache-kafka/README.md b/apache-kafka/README.md new file mode 100644 index 0000000000..5e724f95b6 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/apache-kafka/pom.xml b/apache-kafka/pom.xml new file mode 100644 index 0000000000..41cbab44ed --- /dev/null +++ b/apache-kafka/pom.xml @@ -0,0 +1,181 @@ + + + 4.0.0 + apache-kafka + apache-kafka + + + com.baeldung + parent-modules + 1.0.0-SNAPSHOT + + + + + org.apache.kafka + kafka-clients + ${kafka.version} + + + org.apache.kafka + kafka-streams + ${kafka.version} + + + org.slf4j + slf4j-api + ${slf4j.version} + + + org.slf4j + slf4j-log4j12 + ${slf4j.version} + + + org.apache.flink + flink-connector-kafka-0.11_2.11 + ${flink.version} + + + org.apache.flink + flink-streaming-java_2.11 + ${flink.version} + + + org.apache.flink + flink-core + ${flink.version} + + + commons-logging + commons-logging + + + + + org.apache.flink + flink-java + ${flink.version} + + + commons-logging + commons-logging + + + + + org.apache.flink + flink-test-utils_2.11 + ${flink.version} + test + + + com.google.guava + guava + ${guava.version} + + + org.awaitility + awaitility + ${awaitility.version} + test + + + org.awaitility + awaitility-proxy + ${awaitility.version} + test + + + com.fasterxml.jackson.datatype + jackson-datatype-jsr310 + ${jackson.version} + + + com.fasterxml.jackson.core + jackson-databind + ${jackson.version} + + + org.assertj + assertj-core + ${assertj.version} + test + + + org.testcontainers + kafka + ${testcontainers-kafka.version} + test + + + org.testcontainers + junit-jupiter + ${testcontainers-jupiter.version} + test + + + org.apache.spark + spark-core_2.11 + ${org.apache.spark.spark-core.version} + provided + + + org.apache.spark + spark-sql_2.11 + ${org.apache.spark.spark-core.version} + provided + + + org.apache.spark + spark-graphx_2.11 + ${org.apache.spark.spark-core.version} + provided + + + org.apache.spark + spark-streaming_2.11 + ${org.apache.spark.spark-core.version} + provided + + + org.apache.spark + spark-mllib_2.11 + ${org.apache.spark.spark-core.version} + provided + + + org.apache.spark + spark-streaming-kafka-0-10_2.11 + ${org.apache.spark.spark-core.version} + + + com.datastax.spark + spark-cassandra-connector_2.11 + ${com.datastax.spark.spark-cassandra-connector.version} + + + com.datastax.spark + spark-cassandra-connector-java_2.11 + ${com.datastax.spark.spark-cassandra-connector-java.version} + + + + + 3.6.2 + 1.7.25 + 2.8.0 + 1.15.3 + 1.15.3 + 1.5.0 + 3.0.0 + 29.0-jre + 2.4.8 + 0.8.1-spark3.0-s_2.12 + 2.5.2 + 1.6.0-M1 + + + \ No newline at end of file From 3e41f10f7fdd73894560f96ec58a8cec1416caab Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Sun, 15 Aug 2021 17:32:08 +0530 Subject: [PATCH 10/12] JAVA-6390: main pom changes to add new module --- pom.xml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pom.xml b/pom.xml index abdb851734..803533869f 100644 --- a/pom.xml +++ b/pom.xml @@ -345,6 +345,7 @@ antlr apache-cxf + apache-kafka apache-libraries apache-olingo/olingo2 apache-poi @@ -814,6 +815,7 @@ antlr apache-cxf + apache-kafka apache-libraries apache-olingo/olingo2 apache-poi From 7871c67031fad66a600d7d015e4e7222099925dc Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Mon, 16 Aug 2021 22:12:18 +0530 Subject: [PATCH 11/12] remove empty module --- .../kafkastreams/KafkaStreamsLiveTest.java | 58 ++++++++++--------- libraries-data-3/README.md | 9 --- libraries-data-3/pom.xml | 21 ------- 3 files changed, 30 insertions(+), 58 deletions(-) delete mode 100644 libraries-data-3/README.md delete mode 100644 libraries-data-3/pom.xml diff --git a/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java b/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java index 0b66dd8fec..3b559b619e 100644 --- a/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java +++ b/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java @@ -1,5 +1,9 @@ package com.baeldung.kafkastreams; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Arrays; import java.util.Properties; import java.util.regex.Pattern; @@ -18,59 +22,57 @@ import org.junit.Test; public class KafkaStreamsLiveTest { private String bootstrapServers = "localhost:9092"; + private Path stateDirectory; @Test @Ignore("it needs to have kafka broker running on local") public void shouldTestKafkaStreams() throws InterruptedException { - //given + // given String inputTopic = "inputTopic"; Properties streamsConfiguration = new Properties(); streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-live-test"); streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); - streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); - streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); + streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String() + .getClass() + .getName()); + streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String() + .getClass() + .getName()); streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - // Use a temporary directory for storing state, which will be automatically removed after the test. - // streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath()); - /* - * final StreamsBuilder builder = new StreamsBuilder(); - KStream textLines = builder.stream(wordCountTopic, - Consumed.with(Serdes.String(), Serdes.String())); + // Use a temporary directory for storing state, which will be automatically removed after the test. + try { + this.stateDirectory = Files.createTempDirectory("kafka-streams"); + streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, this.stateDirectory.toAbsolutePath() + .toString()); + } catch (final IOException e) { + throw new UncheckedIOException("Cannot create temporary directory", e); + } - KTable wordCounts = textLines - .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT) - .split("\\W+"))) - .groupBy((key, word) -> word) - .count(Materialized.> as("counts-store")); - */ - //when + // when final StreamsBuilder builder = new StreamsBuilder(); KStream textLines = builder.stream(inputTopic); Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS); - KTable wordCounts = textLines - .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase()))) - .groupBy((key, word) -> word) - .count(); + KTable wordCounts = textLines.flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase()))) + .groupBy((key, word) -> word) + .count(); - wordCounts.toStream().foreach((word, count) -> System.out.println("word: " + word + " -> " + count)); + wordCounts.toStream() + .foreach((word, count) -> System.out.println("word: " + word + " -> " + count)); String outputTopic = "outputTopic"; - //final Serde stringSerde = Serdes.String(); - //final Serde longSerde = Serdes.Long(); - //wordCounts.toStream().to(stringSerde, longSerde, outputTopic); - - wordCounts.toStream().to("outputTopic", - Produced.with(Serdes.String(), Serdes.Long())); + + wordCounts.toStream() + .to(outputTopic, Produced.with(Serdes.String(), Serdes.Long())); final Topology topology = builder.build(); KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration); streams.start(); - //then + // then Thread.sleep(30000); streams.close(); } diff --git a/libraries-data-3/README.md b/libraries-data-3/README.md deleted file mode 100644 index fe856436f1..0000000000 --- a/libraries-data-3/README.md +++ /dev/null @@ -1,9 +0,0 @@ -## Data Libraries - -This module contains articles about libraries for data processing in Java. - -### Relevant articles - - -##### Building the project -You can build the project from the command line using: *mvn clean install*, or in an IDE. \ No newline at end of file diff --git a/libraries-data-3/pom.xml b/libraries-data-3/pom.xml deleted file mode 100644 index 1a95613cb3..0000000000 --- a/libraries-data-3/pom.xml +++ /dev/null @@ -1,21 +0,0 @@ - - - 4.0.0 - libraries-data-3 - libraries-data-3 - - - com.baeldung - parent-modules - 1.0.0-SNAPSHOT - - - - - - - - - \ No newline at end of file From c4011204672a2aabceaa64762e3cf3a81046be93 Mon Sep 17 00:00:00 2001 From: sampadawagde Date: Tue, 17 Aug 2021 14:21:04 +0530 Subject: [PATCH 12/12] JAVA-6390: review corrections --- apache-kafka/pom.xml | 5 ++--- .../src/main/java/com/baeldung/flink/model/InputMessage.java | 5 ++++- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/apache-kafka/pom.xml b/apache-kafka/pom.xml index 41cbab44ed..cda91ed92f 100644 --- a/apache-kafka/pom.xml +++ b/apache-kafka/pom.xml @@ -26,12 +26,12 @@ org.slf4j slf4j-api - ${slf4j.version} + ${org.slf4j.version} org.slf4j slf4j-log4j12 - ${slf4j.version} + ${org.slf4j.version} org.apache.flink @@ -165,7 +165,6 @@ 3.6.2 - 1.7.25 2.8.0 1.15.3 1.15.3 diff --git a/apache-kafka/src/main/java/com/baeldung/flink/model/InputMessage.java b/apache-kafka/src/main/java/com/baeldung/flink/model/InputMessage.java index 9331811b91..d33eb5a9ac 100644 --- a/apache-kafka/src/main/java/com/baeldung/flink/model/InputMessage.java +++ b/apache-kafka/src/main/java/com/baeldung/flink/model/InputMessage.java @@ -61,7 +61,10 @@ public class InputMessage { 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) && Objects.equal(message, message1.message); + return Objects.equal(sender, message1.sender) && + Objects.equal(recipient, message1.recipient) && + Objects.equal(sentAt, message1.sentAt) && + Objects.equal(message, message1.message); } @Override