BAEL-7258: renaming

etrandafir93 · etrandafir93 · commit 40ef28a4497b · 2023-12-09T19:20:16.000+01:00
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
@@ -12,7 +12,7 @@
 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 @@ private static KafkaConsumer<String, String> defaultKafkaConsumer(String boostra
         return new KafkaConsumer<>(props);
     }
 
-    public CustomKafkaListener doForEach(Consumer<String> newConsumer) {
+    public CustomKafkaListener onEach(Consumer<String> newConsumer) {
         recordConsumer = recordConsumer.andThen(newConsumer);
         return this;
     }
@@ -52,10 +52,9 @@ public CustomKafkaListener doForEach(Consumer<String> newConsumer) {
     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/CustomKafkaListenerLiveTest.java b/apache-kafka-2/src/test/java/com/baeldung/kafka/consumer/CustomKafkaListenerLiveTest.java
@@ -7,6 +7,7 @@
 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.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<String> 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 @@ void test() {
             ));
     }
 
-    private void publishArticles(String topic, List<String> articles) {
+    private void publishArticles(String topic, String... articles) {
         try (KafkaProducer<String, String> producer = testKafkaProducer()) {
-            articles.stream()
+            Arrays.stream(articles)
                 .map(article -> new ProducerRecord<String,String>(topic, article))
                 .forEach(producer::send);
         }

Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,7 @@`
`12`	`12`	`import org.apache.kafka.clients.consumer.KafkaConsumer;`
`13`	`13`	`import org.apache.kafka.common.serialization.StringDeserializer;`
`14`	`14`
`15`		`-public class CustomKafkaListener implements Runnable, Closeable {`
	`15`	`+public class CustomKafkaListener implements Runnable, AutoCloseable {`
`16`	`16`
`17`	`17`	`private static final Logger log = Logger.getLogger(CustomKafkaListener.class.getName());`
`18`	`18`
`@@ -43,7 +43,7 @@ private static KafkaConsumer<String, String> defaultKafkaConsumer(String boostra`
`43`	`43`	`return new KafkaConsumer<>(props);`
`44`	`44`	`}`
`45`	`45`
`46`		`- public CustomKafkaListener doForEach(Consumer<String> newConsumer) {`
	`46`	`+ public CustomKafkaListener onEach(Consumer<String> newConsumer) {`
`47`	`47`	`recordConsumer = recordConsumer.andThen(newConsumer);`
`48`	`48`	`return this;`
`49`	`49`	`}`
`@@ -52,10 +52,9 @@ public CustomKafkaListener doForEach(Consumer<String> newConsumer) {`
`52`	`52`	`public void run() {`
`53`	`53`	`running.set(true);`
`54`	`54`	`consumer.subscribe(Arrays.asList(topic));`
`55`		`-`
`56`	`55`	`while (running.get()) {`
`57`	`56`	`consumer.poll(Duration.ofMillis(100))`
`58`		`- .forEach(record -> recordConsumer.accept(record.value()));`
	`57`	`+ .forEach(record -> recordConsumer.accept(record.value()));`
`59`	`58`	`}`
`60`	`59`	`}`
`61`	`60`