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/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/apache-kafka/pom.xml b/apache-kafka/pom.xml new file mode 100644 index 0000000000..cda91ed92f --- /dev/null +++ b/apache-kafka/pom.xml @@ -0,0 +1,180 @@ + + + 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 + ${org.slf4j.version} + + + org.slf4j + slf4j-log4j12 + ${org.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 + 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 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 82% 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..d33eb5a9ac 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,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); } 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-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-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-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() { 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/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/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-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/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-3/src/test/java/com/baeldung/kafka/streams/KafkaStreamsLiveTest.java b/apache-kafka/src/test/java/com/baeldung/kafka/streamsvsconsumer/KafkaStreamsLiveTest.java similarity index 99% rename from libraries-data-3/src/test/java/com/baeldung/kafka/streams/KafkaStreamsLiveTest.java rename to apache-kafka/src/test/java/com/baeldung/kafka/streamsvsconsumer/KafkaStreamsLiveTest.java index 0d4c0606e3..88de6101dc 100644 --- a/libraries-data-3/src/test/java/com/baeldung/kafka/streams/KafkaStreamsLiveTest.java +++ b/apache-kafka/src/test/java/com/baeldung/kafka/streamsvsconsumer/KafkaStreamsLiveTest.java @@ -1,4 +1,4 @@ -package com.baeldung.kafka.streams; +package com.baeldung.kafka.streamsvsconsumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.KafkaProducer; diff --git a/libraries-data/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java b/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java similarity index 53% rename from libraries-data/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java rename to apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java index 32568e9ea5..3b559b619e 100644 --- a/libraries-data/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java +++ b/apache-kafka/src/test/java/com/baeldung/kafkastreams/KafkaStreamsLiveTest.java @@ -1,61 +1,78 @@ 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.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; +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"; + 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()); + 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); + } - //when - KStreamBuilder builder = new KStreamBuilder(); + // 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.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.to(stringSerde, longSerde, outputTopic); - KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration); + 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-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 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 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; - } - } diff --git a/libraries-data-3/README.md b/libraries-data-3/README.md deleted file mode 100644 index d3bfb2c80c..0000000000 --- a/libraries-data-3/README.md +++ /dev/null @@ -1,11 +0,0 @@ -## Data Libraries - -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. diff --git a/libraries-data-3/pom.xml b/libraries-data-3/pom.xml deleted file mode 100644 index 37d5c7ca0d..0000000000 --- a/libraries-data-3/pom.xml +++ /dev/null @@ -1,64 +0,0 @@ - - - 4.0.0 - libraries-data-3 - libraries-data-3 - - - 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.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 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 diff --git a/pom.xml b/pom.xml index f56006713a..059c6dae25 100644 --- a/pom.xml +++ b/pom.xml @@ -345,6 +345,7 @@ antlr apache-cxf + apache-kafka apache-libraries apache-olingo/olingo2 apache-poi @@ -815,6 +816,7 @@ antlr apache-cxf + apache-kafka apache-libraries apache-olingo/olingo2 apache-poi