From 5d13422a53d95116560abe2d48da18e8107d56b3 Mon Sep 17 00:00:00 2001 From: emanueltrandafir1993 Date: Sun, 3 Dec 2023 22:17:16 +0100 Subject: [PATCH 1/4] BAEL-7258: kafka listener without spring --- .../kafka/consumer/CustomKafkaListener.java | 68 +++++++++++++++ .../KafkaListenerWithoutSpringLiveTest.java | 85 +++++++++++++++++++ 2 files changed, 153 insertions(+) create mode 100644 apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java create mode 100644 apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/KafkaListenerWithoutSpringLiveTest.java diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java new file mode 100644 index 0000000000..58987d473b --- /dev/null +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java @@ -0,0 +1,68 @@ +package com.baeldung.kafka.consumer; + +import java.io.Closeable; +import java.time.Duration; +import java.util.Arrays; +import java.util.Properties; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.logging.Logger; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.serialization.StringDeserializer; + +public class CustomKafkaListener implements Runnable, Closeable { + + private static final Logger log = Logger.getLogger(CustomKafkaListener.class.getName()); + + private final String topic; + private final KafkaConsumer consumer; + + private final AtomicBoolean running = new AtomicBoolean(false); + private Consumer recordConsumer; + + + public CustomKafkaListener(String topic, KafkaConsumer consumer) { + this.topic = topic; + this.consumer = consumer; + this.recordConsumer = record -> log.info("received: " + record); + } + + public CustomKafkaListener(String topic, String bootstrapServers) { + this(topic, defaultKafkaConsumer(bootstrapServers)); + } + + private static KafkaConsumer defaultKafkaConsumer(String boostrapServers) { + Properties props = new Properties(); + props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, boostrapServers); + props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test_group_id"); + props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + return new KafkaConsumer<>(props); + } + + public CustomKafkaListener doForEach(Consumer newConsumer) { + recordConsumer = recordConsumer.andThen(newConsumer); + return this; + } + + @Override + public void run() { + running.set(true); + consumer.subscribe(Arrays.asList(topic)); + + while (running.get()) { + consumer.poll(Duration.ofMillis(100)) + .forEach(record -> recordConsumer.accept(record.value())); + } + } + + @Override + public void close() { + running.set(false); + consumer.close(); + } + +} diff --git a/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/KafkaListenerWithoutSpringLiveTest.java b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/KafkaListenerWithoutSpringLiveTest.java new file mode 100644 index 0000000000..213344d0a8 --- /dev/null +++ b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/KafkaListenerWithoutSpringLiveTest.java @@ -0,0 +1,85 @@ +package com.baeldung.kafka.consumer; + +import static java.time.Duration.ofMillis; +import static java.time.Duration.ofSeconds; +import static java.util.Arrays.asList; +import static org.assertj.core.api.Assertions.assertThat; +import static org.testcontainers.shaded.org.awaitility.Awaitility.await; + +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.CompletableFuture; + +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.KafkaContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.shaded.org.awaitility.Awaitility; +import org.testcontainers.utility.DockerImageName; + +@Testcontainers +class KafkaListenerWithoutSpringLiveTest { + + @Container + private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest")); + + static { + Awaitility.setDefaultTimeout(ofSeconds(1L)); + Awaitility.setDefaultPollInterval(ofMillis(50L)); + } + + @Test + void test() { + // given + String topic = "baeldung.articles.published"; + String bootstrapServers = KAFKA_CONTAINER.getBootstrapServers(); + List consumedMessages = new ArrayList<>(); + + // when + try (CustomKafkaListener listener = new CustomKafkaListener(topic, bootstrapServers)) { + CompletableFuture.runAsync(() -> + listener.doForEach(consumedMessages::add).run() + ); + } + // and + publishArticles(topic, asList( + "Introduction to Kafka", + "Kotlin for Java Developers", + "Reactive Spring Boot", + "Deploying Spring Boot Applications", + "Spring Security" + )); + + // then + await().untilAsserted(() -> assertThat(consumedMessages) + .containsExactlyInAnyOrder( + "Introduction to Kafka", + "Kotlin for Java Developers", + "Reactive Spring Boot", + "Deploying Spring Boot Applications", + "Spring Security" + )); + } + + private void publishArticles(String topic, List articles) { + try (KafkaProducer producer = testKafkaProducer()) { + articles.stream() + .map(article -> new ProducerRecord(topic, article)) + .forEach(producer::send); + } + } + + private static KafkaProducer testKafkaProducer() { + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers()); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + return new KafkaProducer<>(props); + } + +} \ No newline at end of file From 40ef28a4497b1c4b02426a108eb0f2cd899db9ef Mon Sep 17 00:00:00 2001 From: emanueltrandafir1993 Date: Sat, 9 Dec 2023 19:20:16 +0100 Subject: [PATCH 2/4] BAEL-7258: renaming --- .../kafka/consumer/CustomKafkaListener.java | 7 +++--- ....java => CustomKafkaListenerLiveTest.java} | 23 +++++++++---------- 2 files changed, 14 insertions(+), 16 deletions(-) rename apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/{KafkaListenerWithoutSpringLiveTest.java => CustomKafkaListenerLiveTest.java} (84%) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java index 58987d473b..b432c2ba55 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java @@ -12,7 +12,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; -public class CustomKafkaListener implements Runnable, Closeable { +public class CustomKafkaListener implements Runnable, AutoCloseable { private static final Logger log = Logger.getLogger(CustomKafkaListener.class.getName()); @@ -43,7 +43,7 @@ public class CustomKafkaListener implements Runnable, Closeable { return new KafkaConsumer<>(props); } - public CustomKafkaListener doForEach(Consumer newConsumer) { + public CustomKafkaListener onEach(Consumer newConsumer) { recordConsumer = recordConsumer.andThen(newConsumer); return this; } @@ -52,10 +52,9 @@ public class CustomKafkaListener implements Runnable, Closeable { public void run() { running.set(true); consumer.subscribe(Arrays.asList(topic)); - while (running.get()) { consumer.poll(Duration.ofMillis(100)) - .forEach(record -> recordConsumer.accept(record.value())); + .forEach(record -> recordConsumer.accept(record.value())); } } diff --git a/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/KafkaListenerWithoutSpringLiveTest.java b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java similarity index 84% rename from apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/KafkaListenerWithoutSpringLiveTest.java rename to apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java index 213344d0a8..dbe7483dcd 100644 --- a/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/KafkaListenerWithoutSpringLiveTest.java +++ b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java @@ -7,6 +7,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.testcontainers.shaded.org.awaitility.Awaitility.await; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Properties; import java.util.concurrent.CompletableFuture; @@ -23,7 +24,7 @@ import org.testcontainers.shaded.org.awaitility.Awaitility; import org.testcontainers.utility.DockerImageName; @Testcontainers -class KafkaListenerWithoutSpringLiveTest { +class CustomKafkaListenerLiveTest { @Container private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest")); @@ -34,30 +35,28 @@ class KafkaListenerWithoutSpringLiveTest { } @Test - void test() { + void givenANewCustomKafkaListener_thenConsumesAllMessages() { // given String topic = "baeldung.articles.published"; String bootstrapServers = KAFKA_CONTAINER.getBootstrapServers(); List consumedMessages = new ArrayList<>(); // when - try (CustomKafkaListener listener = new CustomKafkaListener(topic, bootstrapServers)) { - CompletableFuture.runAsync(() -> - listener.doForEach(consumedMessages::add).run() - ); + try (CustomKafkaListener listener = new CustomKafkaListener(topic, bootstrapServers).onEach(consumedMessages::add)) { + CompletableFuture.runAsync(listener); } // and - publishArticles(topic, asList( + publishArticles(topic, "Introduction to Kafka", "Kotlin for Java Developers", "Reactive Spring Boot", "Deploying Spring Boot Applications", "Spring Security" - )); + ); // then - await().untilAsserted(() -> assertThat(consumedMessages) - .containsExactlyInAnyOrder( + await().untilAsserted(() -> + assertThat(consumedMessages).containsExactlyInAnyOrder( "Introduction to Kafka", "Kotlin for Java Developers", "Reactive Spring Boot", @@ -66,9 +65,9 @@ class KafkaListenerWithoutSpringLiveTest { )); } - private void publishArticles(String topic, List articles) { + private void publishArticles(String topic, String... articles) { try (KafkaProducer producer = testKafkaProducer()) { - articles.stream() + Arrays.stream(articles) .map(article -> new ProducerRecord(topic, article)) .forEach(producer::send); } From 0203ff1f0a4e2b81ea507a636dc1a54d82360fd9 Mon Sep 17 00:00:00 2001 From: "emanuel.trandafir" Date: Fri, 15 Dec 2023 12:44:12 +0100 Subject: [PATCH 3/4] BAEL-7258: increase pom versions --- apache-kafka-2/pom.xml | 4 +- .../kafka/consumer/CustomKafkaListener.java | 15 +----- .../consumer/CustomKafkaListenerLiveTest.java | 53 ++++++++++++------- 3 files changed, 38 insertions(+), 34 deletions(-) diff --git a/apache-kafka-2/pom.xml b/apache-kafka-2/pom.xml index cd21a60e14..53217731c7 100644 --- a/apache-kafka-2/pom.xml +++ b/apache-kafka-2/pom.xml @@ -61,9 +61,9 @@ 5.7.0 - 2.8.0 + 3.6.1 1.19.3 - 1.15.3 + 1.19.3 2.15.2 diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java index b432c2ba55..f4cefb21eb 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/consumer/CustomKafkaListener.java @@ -12,14 +12,10 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; -public class CustomKafkaListener implements Runnable, AutoCloseable { - +public class CustomKafkaListener implements Runnable { private static final Logger log = Logger.getLogger(CustomKafkaListener.class.getName()); - private final String topic; private final KafkaConsumer consumer; - - private final AtomicBoolean running = new AtomicBoolean(false); private Consumer recordConsumer; @@ -50,18 +46,11 @@ public class CustomKafkaListener implements Runnable, AutoCloseable { @Override public void run() { - running.set(true); consumer.subscribe(Arrays.asList(topic)); - while (running.get()) { + while (true) { consumer.poll(Duration.ofMillis(100)) .forEach(record -> recordConsumer.accept(record.value())); } } - @Override - public void close() { - running.set(false); - consumer.close(); - } - } diff --git a/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java index dbe7483dcd..cba0aacce6 100644 --- a/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java +++ b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java @@ -4,18 +4,32 @@ import static java.time.Duration.ofMillis; import static java.time.Duration.ofSeconds; import static java.util.Arrays.asList; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.testcontainers.shaded.org.awaitility.Awaitility.await; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import java.util.Properties; +import java.time.Duration; +import java.util.*; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.testcontainers.containers.KafkaContainer; import org.testcontainers.junit.jupiter.Container; @@ -30,7 +44,7 @@ class CustomKafkaListenerLiveTest { private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest")); static { - Awaitility.setDefaultTimeout(ofSeconds(1L)); + Awaitility.setDefaultTimeout(ofSeconds(5L)); Awaitility.setDefaultPollInterval(ofMillis(50L)); } @@ -42,33 +56,34 @@ class CustomKafkaListenerLiveTest { List consumedMessages = new ArrayList<>(); // when - try (CustomKafkaListener listener = new CustomKafkaListener(topic, bootstrapServers).onEach(consumedMessages::add)) { - CompletableFuture.runAsync(listener); - } + CustomKafkaListener listener = new CustomKafkaListener(topic, bootstrapServers).onEach(consumedMessages::add); + CompletableFuture.runAsync(listener); + // and publishArticles(topic, + "Introduction to Kafka", + "Kotlin for Java Developers", + "Reactive Spring Boot", + "Deploying Spring Boot Applications", + "Spring Security" + ); + + // then + await().untilAsserted(() -> + assertThat(consumedMessages).containsExactlyInAnyOrder( "Introduction to Kafka", "Kotlin for Java Developers", "Reactive Spring Boot", "Deploying Spring Boot Applications", "Spring Security" - ); + )); - // then - await().untilAsserted(() -> - assertThat(consumedMessages).containsExactlyInAnyOrder( - "Introduction to Kafka", - "Kotlin for Java Developers", - "Reactive Spring Boot", - "Deploying Spring Boot Applications", - "Spring Security" - )); } private void publishArticles(String topic, String... articles) { try (KafkaProducer producer = testKafkaProducer()) { Arrays.stream(articles) - .map(article -> new ProducerRecord(topic, article)) + .map(article -> new ProducerRecord<>(topic, "key-1", article)) .forEach(producer::send); } } From 89817e2cda14014914ebcc34c48ca1b6d25ed085 Mon Sep 17 00:00:00 2001 From: "emanuel.trandafir" Date: Fri, 15 Dec 2023 12:47:19 +0100 Subject: [PATCH 4/4] MKT-7258: optimize imports --- .../consumer/CustomKafkaListenerLiveTest.java | 25 +++---------------- 1 file changed, 4 insertions(+), 21 deletions(-) diff --git a/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java index cba0aacce6..201a72114d 100644 --- a/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java +++ b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java @@ -2,34 +2,19 @@ package com.baeldung.kafka.consumer; import static java.time.Duration.ofMillis; import static java.time.Duration.ofSeconds; -import static java.util.Arrays.asList; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.testcontainers.shaded.org.awaitility.Awaitility.await; -import java.time.Duration; -import java.util.*; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Properties; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import org.apache.kafka.clients.admin.Admin; -import org.apache.kafka.clients.admin.AdminClientConfig; -import org.apache.kafka.clients.admin.NewTopic; -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; -import org.apache.kafka.common.header.Header; -import org.apache.kafka.common.header.Headers; -import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.testcontainers.containers.KafkaContainer; import org.testcontainers.junit.jupiter.Container; @@ -77,7 +62,6 @@ class CustomKafkaListenerLiveTest { "Deploying Spring Boot Applications", "Spring Security" )); - } private void publishArticles(String topic, String... articles) { @@ -95,5 +79,4 @@ class CustomKafkaListenerLiveTest { props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); return new KafkaProducer<>(props); } - } \ No newline at end of file