From f617d7985d2a461daf0e4c2477a9ebbbea8bc9f2 Mon Sep 17 00:00:00 2001 From: s9m33r Date: Sun, 28 Jan 2024 17:41:58 +0530 Subject: [PATCH 1/8] BAEL-7351 Demo code --- .../spring/kafka/groupId/KafkaConfig.java | 47 +++++++++++++++++++ .../kafka/groupId/KafkaErrorHandler.java | 37 +++++++++++++++ .../baeldung/spring/kafka/groupId/Main.java | 14 ++++++ .../spring/kafka/groupId/MyKafkaConsumer.java | 41 ++++++++++++++++ .../spring/kafka/groupId/MyKafkaProducer.java | 24 ++++++++++ .../spring/kafka/groupId/MainLiveTest.java | 44 +++++++++++++++++ 6 files changed, 207 insertions(+) create mode 100644 spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java create mode 100644 spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java create mode 100644 spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java create mode 100644 spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java create mode 100644 spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java create mode 100644 spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java new file mode 100644 index 0000000000..b8d7c42ec9 --- /dev/null +++ b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java @@ -0,0 +1,47 @@ +package com.baeldung.spring.kafka.groupId; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.listener.CommonErrorHandler; + +@EnableKafka +@Configuration +class KafkaConfig { + + @Bean + ConsumerFactory consumerFactory(@Value("${spring.kafka.bootstrap-servers:localhost:9092}") String bootstrapServers) { + Map props = new HashMap<>(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ConsumerConfig.GROUP_ID_CONFIG, "${kafka.consumer.groupId:test-consumer-group}"); + props.put(ConsumerConfig.CLIENT_ID_CONFIG, "rex"); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new StringDeserializer()); + } + + @Bean + ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory consumerFactory, + CommonErrorHandler commonErrorHandler) { + var factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory); + factory.setCommonErrorHandler(commonErrorHandler); + return factory; + } + + @Bean + CommonErrorHandler kafkaErrorHandler() { + return new KafkaErrorHandler(); + } + +} \ No newline at end of file diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java new file mode 100644 index 0000000000..a30254ea47 --- /dev/null +++ b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java @@ -0,0 +1,37 @@ +package com.baeldung.spring.kafka.groupId; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.errors.RecordDeserializationException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.listener.CommonErrorHandler; +import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.lang.NonNull; + +class KafkaErrorHandler implements CommonErrorHandler { + + private static final Logger log = LoggerFactory.getLogger(KafkaErrorHandler.class); + + @Override + public void handleRecord(@NonNull Exception exception, @NonNull ConsumerRecord record, @NonNull Consumer consumer, + @NonNull MessageListenerContainer container) { + handle(exception, consumer); + } + + @Override + public void handleOtherException(@NonNull Exception exception, @NonNull Consumer consumer, @NonNull MessageListenerContainer container, + boolean batchListener) { + handle(exception, consumer); + } + + private void handle(Exception exception, Consumer consumer) { + log.error("Exception thrown", exception); + if (exception instanceof RecordDeserializationException ex) { + consumer.seek(ex.topicPartition(), ex.offset() + 1L); + consumer.commitSync(); + } else { + log.error("Exception not handled", exception); + } + } +} \ No newline at end of file diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java new file mode 100644 index 0000000000..1045bc5192 --- /dev/null +++ b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java @@ -0,0 +1,14 @@ +package com.baeldung.spring.kafka.groupId; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.ComponentScan; + +@SpringBootApplication +@ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") +public class Main { + + public static void main(String[] args) { + SpringApplication.run(Main.class, args); + } +} diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java new file mode 100644 index 0000000000..e17d168338 --- /dev/null +++ b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java @@ -0,0 +1,41 @@ +package com.baeldung.spring.kafka.groupId; + +import java.util.concurrent.CountDownLatch; + +import org.apache.kafka.clients.consumer.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Service; + +@Service +public class MyKafkaConsumer { + + private static final Logger LOGGER = LoggerFactory.getLogger(MyKafkaConsumer.class); + + private CountDownLatch latch = new CountDownLatch(1); + + private String payload; + + @KafkaListener(topics = "${kafka.topic.name:test-topic}", groupId = "${kafka.consumer.groupId:test-consumer-group}", concurrency = "4") + public void receive(@Payload String payload, Consumer consumer) { + LOGGER.info("Consumer='{}' received payload='{}'", consumer.groupMetadata() + .memberId(), payload); + this.payload = payload; + + latch.countDown(); + } + + public CountDownLatch getLatch() { + return latch; + } + + public void resetLatch() { + latch = new CountDownLatch(1); + } + + public String getPayload() { + return payload; + } +} diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java new file mode 100644 index 0000000000..c1d9a05d60 --- /dev/null +++ b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java @@ -0,0 +1,24 @@ +package com.baeldung.spring.kafka.groupId; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.stereotype.Service; + +@Service +public class MyKafkaProducer { + + private static final Logger LOGGER = LoggerFactory.getLogger(MyKafkaProducer.class); + + @Value("${kafka.topic.name:test-topic}") + private String topic; + @Autowired + private KafkaTemplate kafkaTemplate; + + public void send(String payload) { + LOGGER.info("Sending payload='{}' to topic='{}'", payload, topic); + kafkaTemplate.send(topic, payload); + } +} diff --git a/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java b/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java new file mode 100644 index 0000000000..97b056db73 --- /dev/null +++ b/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java @@ -0,0 +1,44 @@ +package com.baeldung.spring.kafka.groupId; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.test.annotation.DirtiesContext; + +@SpringBootTest(classes = Main.class) +@ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") +@DirtiesContext +@EmbeddedKafka(partitions = 4, topics = { "${kafka.topic.name:test-topic}" }, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" }) +public class MainLiveTest { + + @Autowired + private MyKafkaConsumer consumer; + @Autowired + private MyKafkaProducer producer; + + @BeforeEach + void setup() { + consumer.resetLatch(); + } + + @Test + public void givenEmbeddedKafkaBroker_whenSendingWithSimpleProducer_thenMessageReceived() throws Exception { + String data = "Test 123..."; + + producer.send(data); + + boolean messageConsumed = consumer.getLatch() + .await(10, TimeUnit.SECONDS); + assertTrue(messageConsumed); + assertThat(consumer.getPayload(), containsString(data)); + } +} \ No newline at end of file From ca9ebc37cfef17d3bd80c459610891ead94a71b1 Mon Sep 17 00:00:00 2001 From: s9m33r Date: Sun, 28 Jan 2024 19:59:17 +0530 Subject: [PATCH 2/8] BAEL-7351 added new lines --- .../java/com/baeldung/spring/kafka/groupId/KafkaConfig.java | 2 +- .../com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java | 2 +- .../java/com/baeldung/spring/kafka/groupId/MainLiveTest.java | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java index b8d7c42ec9..6514816424 100644 --- a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java +++ b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java @@ -44,4 +44,4 @@ class KafkaConfig { return new KafkaErrorHandler(); } -} \ No newline at end of file +} diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java index a30254ea47..00cc795272 100644 --- a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java +++ b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java @@ -34,4 +34,4 @@ class KafkaErrorHandler implements CommonErrorHandler { log.error("Exception not handled", exception); } } -} \ No newline at end of file +} diff --git a/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java b/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java index 97b056db73..c7e78f6e58 100644 --- a/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java +++ b/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java @@ -41,4 +41,4 @@ public class MainLiveTest { assertTrue(messageConsumed); assertThat(consumer.getPayload(), containsString(data)); } -} \ No newline at end of file +} From d8e035da08782a110ca8e75e7bdc96c90d4ae7fa Mon Sep 17 00:00:00 2001 From: s9m33r Date: Wed, 31 Jan 2024 00:40:41 +0530 Subject: [PATCH 3/8] BAEL-7351 moving to a new module --- apache-kafka-3/.gitignore | 33 +++++++ apache-kafka-3/pom.xml | 50 +++++++++++ .../apachekafka3}/groupId/KafkaConfig.java | 2 +- .../groupId/KafkaErrorHandler.java | 9 +- .../baeldung/apachekafka3}/groupId/Main.java | 28 +++--- .../groupId/MyKafkaConsumer.java | 82 ++++++++--------- .../groupId/MyKafkaProducer.java | 48 +++++----- .../src/main/resources/application.properties | 1 + .../apachekafka3}/groupId/MainLiveTest.java | 88 +++++++++---------- pom.xml | 3 +- 10 files changed, 211 insertions(+), 133 deletions(-) create mode 100644 apache-kafka-3/.gitignore create mode 100644 apache-kafka-3/pom.xml rename {spring-kafka-3/src/main/java/com/baeldung/spring/kafka => apache-kafka-3/src/main/java/com/baeldung/apachekafka3}/groupId/KafkaConfig.java (97%) rename {spring-kafka-3/src/main/java/com/baeldung/spring/kafka => apache-kafka-3/src/main/java/com/baeldung/apachekafka3}/groupId/KafkaErrorHandler.java (76%) rename {spring-kafka-3/src/main/java/com/baeldung/spring/kafka => apache-kafka-3/src/main/java/com/baeldung/apachekafka3}/groupId/Main.java (73%) rename {spring-kafka-3/src/main/java/com/baeldung/spring/kafka => apache-kafka-3/src/main/java/com/baeldung/apachekafka3}/groupId/MyKafkaConsumer.java (82%) rename {spring-kafka-3/src/main/java/com/baeldung/spring/kafka => apache-kafka-3/src/main/java/com/baeldung/apachekafka3}/groupId/MyKafkaProducer.java (91%) create mode 100644 apache-kafka-3/src/main/resources/application.properties rename {spring-kafka-3/src/test/java/com/baeldung/spring/kafka => apache-kafka-3/src/test/java/com/baeldung/apachekafka3}/groupId/MainLiveTest.java (90%) diff --git a/apache-kafka-3/.gitignore b/apache-kafka-3/.gitignore new file mode 100644 index 0000000000..549e00a2a9 --- /dev/null +++ b/apache-kafka-3/.gitignore @@ -0,0 +1,33 @@ +HELP.md +target/ +!.mvn/wrapper/maven-wrapper.jar +!**/src/main/**/target/ +!**/src/test/**/target/ + +### STS ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache + +### IntelliJ IDEA ### +.idea +*.iws +*.iml +*.ipr + +### NetBeans ### +/nbproject/private/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ +build/ +!**/src/main/**/build/ +!**/src/test/**/build/ + +### VS Code ### +.vscode/ diff --git a/apache-kafka-3/pom.xml b/apache-kafka-3/pom.xml new file mode 100644 index 0000000000..9466068482 --- /dev/null +++ b/apache-kafka-3/pom.xml @@ -0,0 +1,50 @@ + + + 4.0.0 + + + com.baeldung + parent-boot-2 + 0.0.1-SNAPSHOT + ../parent-boot-2 + + + apache-kafka-3 + Apache Kafka 3 + Third module for Apache Kafka related articles + + 17 + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.kafka + spring-kafka + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.kafka + spring-kafka-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaConfig.java similarity index 97% rename from spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java rename to apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaConfig.java index 6514816424..da9a41f425 100644 --- a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java +++ b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaConfig.java @@ -1,4 +1,4 @@ -package com.baeldung.spring.kafka.groupId; +package com.baeldung.apachekafka3.groupId; import java.util.HashMap; import java.util.Map; diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaErrorHandler.java similarity index 76% rename from spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java rename to apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaErrorHandler.java index 00cc795272..3acd18a452 100644 --- a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java +++ b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaErrorHandler.java @@ -1,7 +1,6 @@ -package com.baeldung.spring.kafka.groupId; +package com.baeldung.apachekafka3.groupId; import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.errors.RecordDeserializationException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -13,12 +12,6 @@ class KafkaErrorHandler implements CommonErrorHandler { private static final Logger log = LoggerFactory.getLogger(KafkaErrorHandler.class); - @Override - public void handleRecord(@NonNull Exception exception, @NonNull ConsumerRecord record, @NonNull Consumer consumer, - @NonNull MessageListenerContainer container) { - handle(exception, consumer); - } - @Override public void handleOtherException(@NonNull Exception exception, @NonNull Consumer consumer, @NonNull MessageListenerContainer container, boolean batchListener) { diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/Main.java similarity index 73% rename from spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java rename to apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/Main.java index 1045bc5192..e38699270f 100644 --- a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java +++ b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/Main.java @@ -1,14 +1,14 @@ -package com.baeldung.spring.kafka.groupId; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.context.annotation.ComponentScan; - -@SpringBootApplication -@ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") -public class Main { - - public static void main(String[] args) { - SpringApplication.run(Main.class, args); - } -} +package com.baeldung.apachekafka3.groupId; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.ComponentScan; + +@SpringBootApplication +@ComponentScan(basePackages = "com.baeldung.apachekafka3.groupId") +public class Main { + + public static void main(String[] args) { + SpringApplication.run(Main.class, args); + } +} diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaConsumer.java similarity index 82% rename from spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java rename to apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaConsumer.java index e17d168338..e3c7220479 100644 --- a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java +++ b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaConsumer.java @@ -1,41 +1,41 @@ -package com.baeldung.spring.kafka.groupId; - -import java.util.concurrent.CountDownLatch; - -import org.apache.kafka.clients.consumer.Consumer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.kafka.annotation.KafkaListener; -import org.springframework.messaging.handler.annotation.Payload; -import org.springframework.stereotype.Service; - -@Service -public class MyKafkaConsumer { - - private static final Logger LOGGER = LoggerFactory.getLogger(MyKafkaConsumer.class); - - private CountDownLatch latch = new CountDownLatch(1); - - private String payload; - - @KafkaListener(topics = "${kafka.topic.name:test-topic}", groupId = "${kafka.consumer.groupId:test-consumer-group}", concurrency = "4") - public void receive(@Payload String payload, Consumer consumer) { - LOGGER.info("Consumer='{}' received payload='{}'", consumer.groupMetadata() - .memberId(), payload); - this.payload = payload; - - latch.countDown(); - } - - public CountDownLatch getLatch() { - return latch; - } - - public void resetLatch() { - latch = new CountDownLatch(1); - } - - public String getPayload() { - return payload; - } -} +package com.baeldung.apachekafka3.groupId; + +import java.util.concurrent.CountDownLatch; + +import org.apache.kafka.clients.consumer.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Service; + +@Service +public class MyKafkaConsumer { + + private static final Logger LOGGER = LoggerFactory.getLogger(MyKafkaConsumer.class); + + private CountDownLatch latch = new CountDownLatch(1); + + private String payload; + + @KafkaListener(topics = "${kafka.topic.name:test-topic}", clientIdPrefix = "neo", groupId = "${kafka.consumer.groupId:test-consumer-group}", concurrency = "4") + public void receive(@Payload String payload, Consumer consumer) { + LOGGER.info("Consumer='{}' received payload='{}'", consumer.groupMetadata() + .memberId(), payload); + this.payload = payload; + + latch.countDown(); + } + + public CountDownLatch getLatch() { + return latch; + } + + public void resetLatch() { + latch = new CountDownLatch(1); + } + + public String getPayload() { + return payload; + } +} diff --git a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaProducer.java similarity index 91% rename from spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java rename to apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaProducer.java index c1d9a05d60..8707ce800d 100644 --- a/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java +++ b/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaProducer.java @@ -1,24 +1,24 @@ -package com.baeldung.spring.kafka.groupId; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.stereotype.Service; - -@Service -public class MyKafkaProducer { - - private static final Logger LOGGER = LoggerFactory.getLogger(MyKafkaProducer.class); - - @Value("${kafka.topic.name:test-topic}") - private String topic; - @Autowired - private KafkaTemplate kafkaTemplate; - - public void send(String payload) { - LOGGER.info("Sending payload='{}' to topic='{}'", payload, topic); - kafkaTemplate.send(topic, payload); - } -} +package com.baeldung.apachekafka3.groupId; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.stereotype.Service; + +@Service +public class MyKafkaProducer { + + private static final Logger LOGGER = LoggerFactory.getLogger(MyKafkaProducer.class); + + @Value("${kafka.topic.name:test-topic}") + private String topic; + @Autowired + private KafkaTemplate kafkaTemplate; + + public void send(String payload) { + LOGGER.info("Sending payload='{}' to topic='{}'", payload, topic); + kafkaTemplate.send(topic, payload); + } +} diff --git a/apache-kafka-3/src/main/resources/application.properties b/apache-kafka-3/src/main/resources/application.properties new file mode 100644 index 0000000000..0fce641151 --- /dev/null +++ b/apache-kafka-3/src/main/resources/application.properties @@ -0,0 +1 @@ +kafka.topic.name=test-topic diff --git a/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java b/apache-kafka-3/src/test/java/com/baeldung/apachekafka3/groupId/MainLiveTest.java similarity index 90% rename from spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java rename to apache-kafka-3/src/test/java/com/baeldung/apachekafka3/groupId/MainLiveTest.java index c7e78f6e58..e10e41838a 100644 --- a/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java +++ b/apache-kafka-3/src/test/java/com/baeldung/apachekafka3/groupId/MainLiveTest.java @@ -1,44 +1,44 @@ -package com.baeldung.spring.kafka.groupId; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.containsString; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.util.concurrent.TimeUnit; - -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.context.annotation.ComponentScan; -import org.springframework.kafka.test.context.EmbeddedKafka; -import org.springframework.test.annotation.DirtiesContext; - -@SpringBootTest(classes = Main.class) -@ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") -@DirtiesContext -@EmbeddedKafka(partitions = 4, topics = { "${kafka.topic.name:test-topic}" }, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" }) -public class MainLiveTest { - - @Autowired - private MyKafkaConsumer consumer; - @Autowired - private MyKafkaProducer producer; - - @BeforeEach - void setup() { - consumer.resetLatch(); - } - - @Test - public void givenEmbeddedKafkaBroker_whenSendingWithSimpleProducer_thenMessageReceived() throws Exception { - String data = "Test 123..."; - - producer.send(data); - - boolean messageConsumed = consumer.getLatch() - .await(10, TimeUnit.SECONDS); - assertTrue(messageConsumed); - assertThat(consumer.getPayload(), containsString(data)); - } -} +package com.baeldung.apachekafka3.groupId; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.test.annotation.DirtiesContext; + +@SpringBootTest(classes = Main.class) +@ComponentScan(basePackages = "com.baeldung.apachekafka3.groupId") +@DirtiesContext +@EmbeddedKafka(partitions = 4, topics = { "${kafka.topic.name:test-topic}" }, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" }) +public class MainLiveTest { + + @Autowired + private MyKafkaConsumer consumer; + @Autowired + private MyKafkaProducer producer; + + @BeforeEach + void setup() { + consumer.resetLatch(); + } + + @Test + public void givenEmbeddedKafkaBroker_whenSendingWithSimpleProducer_thenMessageReceived() throws Exception { + String data = "Test 123..."; + + producer.send(data); + + boolean messageConsumed = consumer.getLatch() + .await(10, TimeUnit.SECONDS); + assertTrue(messageConsumed); + assertThat(consumer.getPayload(), containsString(data)); + } +} diff --git a/pom.xml b/pom.xml index 030b8e1843..3a3ba46d48 100644 --- a/pom.xml +++ b/pom.xml @@ -670,8 +670,9 @@ apache-httpclient-2 apache-httpclient4 apache-httpclient - apache-kafka-2 apache-kafka + apache-kafka-2 + apache-kafka-3 apache-libraries-2 apache-libraries apache-olingo From a516aca5c49b594f3d3c4303119a4f2eb8206795 Mon Sep 17 00:00:00 2001 From: s9m33r Date: Wed, 31 Jan 2024 21:52:54 +0530 Subject: [PATCH 4/8] BAEL-7351 moving demo code to a spring-kafka-2 module --- apache-kafka-3/.gitignore | 33 ------------ apache-kafka-3/pom.xml | 50 ------------------- .../src/main/resources/application.properties | 1 - pom.xml | 1 - .../spring/kafka}/groupId/KafkaConfig.java | 2 +- .../kafka}/groupId/KafkaErrorHandler.java | 2 +- .../baeldung/spring/kafka}/groupId/Main.java | 4 +- .../kafka}/groupId/MyKafkaConsumer.java | 2 +- .../kafka}/groupId/MyKafkaProducer.java | 2 +- .../spring/kafka}/groupId/MainLiveTest.java | 4 +- 10 files changed, 8 insertions(+), 93 deletions(-) delete mode 100644 apache-kafka-3/.gitignore delete mode 100644 apache-kafka-3/pom.xml delete mode 100644 apache-kafka-3/src/main/resources/application.properties rename {apache-kafka-3/src/main/java/com/baeldung/apachekafka3 => spring-kafka-2/src/main/java/com/baeldung/spring/kafka}/groupId/KafkaConfig.java (97%) rename {apache-kafka-3/src/main/java/com/baeldung/apachekafka3 => spring-kafka-2/src/main/java/com/baeldung/spring/kafka}/groupId/KafkaErrorHandler.java (96%) rename {apache-kafka-3/src/main/java/com/baeldung/apachekafka3 => spring-kafka-2/src/main/java/com/baeldung/spring/kafka}/groupId/Main.java (75%) rename {apache-kafka-3/src/main/java/com/baeldung/apachekafka3 => spring-kafka-2/src/main/java/com/baeldung/spring/kafka}/groupId/MyKafkaConsumer.java (96%) rename {apache-kafka-3/src/main/java/com/baeldung/apachekafka3 => spring-kafka-2/src/main/java/com/baeldung/spring/kafka}/groupId/MyKafkaProducer.java (94%) rename {apache-kafka-3/src/test/java/com/baeldung/apachekafka3 => spring-kafka-2/src/test/java/com/baeldung/spring/kafka}/groupId/MainLiveTest.java (92%) diff --git a/apache-kafka-3/.gitignore b/apache-kafka-3/.gitignore deleted file mode 100644 index 549e00a2a9..0000000000 --- a/apache-kafka-3/.gitignore +++ /dev/null @@ -1,33 +0,0 @@ -HELP.md -target/ -!.mvn/wrapper/maven-wrapper.jar -!**/src/main/**/target/ -!**/src/test/**/target/ - -### STS ### -.apt_generated -.classpath -.factorypath -.project -.settings -.springBeans -.sts4-cache - -### IntelliJ IDEA ### -.idea -*.iws -*.iml -*.ipr - -### NetBeans ### -/nbproject/private/ -/nbbuild/ -/dist/ -/nbdist/ -/.nb-gradle/ -build/ -!**/src/main/**/build/ -!**/src/test/**/build/ - -### VS Code ### -.vscode/ diff --git a/apache-kafka-3/pom.xml b/apache-kafka-3/pom.xml deleted file mode 100644 index 9466068482..0000000000 --- a/apache-kafka-3/pom.xml +++ /dev/null @@ -1,50 +0,0 @@ - - - 4.0.0 - - - com.baeldung - parent-boot-2 - 0.0.1-SNAPSHOT - ../parent-boot-2 - - - apache-kafka-3 - Apache Kafka 3 - Third module for Apache Kafka related articles - - 17 - - - - org.springframework.boot - spring-boot-starter - - - org.springframework.kafka - spring-kafka - - - - org.springframework.boot - spring-boot-starter-test - test - - - org.springframework.kafka - spring-kafka-test - test - - - - - - - org.springframework.boot - spring-boot-maven-plugin - - - - - diff --git a/apache-kafka-3/src/main/resources/application.properties b/apache-kafka-3/src/main/resources/application.properties deleted file mode 100644 index 0fce641151..0000000000 --- a/apache-kafka-3/src/main/resources/application.properties +++ /dev/null @@ -1 +0,0 @@ -kafka.topic.name=test-topic diff --git a/pom.xml b/pom.xml index 3a3ba46d48..330035020f 100644 --- a/pom.xml +++ b/pom.xml @@ -672,7 +672,6 @@ apache-httpclient apache-kafka apache-kafka-2 - apache-kafka-3 apache-libraries-2 apache-libraries apache-olingo diff --git a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaConfig.java b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java similarity index 97% rename from apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaConfig.java rename to spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java index da9a41f425..6514816424 100644 --- a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaConfig.java +++ b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java @@ -1,4 +1,4 @@ -package com.baeldung.apachekafka3.groupId; +package com.baeldung.spring.kafka.groupId; import java.util.HashMap; import java.util.Map; diff --git a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaErrorHandler.java b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java similarity index 96% rename from apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaErrorHandler.java rename to spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java index 3acd18a452..c334925141 100644 --- a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/KafkaErrorHandler.java +++ b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java @@ -1,4 +1,4 @@ -package com.baeldung.apachekafka3.groupId; +package com.baeldung.spring.kafka.groupId; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.common.errors.RecordDeserializationException; diff --git a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/Main.java b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/Main.java similarity index 75% rename from apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/Main.java rename to spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/Main.java index e38699270f..589b20c68f 100644 --- a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/Main.java +++ b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/Main.java @@ -1,11 +1,11 @@ -package com.baeldung.apachekafka3.groupId; +package com.baeldung.spring.kafka.groupId; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.ComponentScan; @SpringBootApplication -@ComponentScan(basePackages = "com.baeldung.apachekafka3.groupId") +@ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") public class Main { public static void main(String[] args) { diff --git a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaConsumer.java b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java similarity index 96% rename from apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaConsumer.java rename to spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java index e3c7220479..f26b605a7b 100644 --- a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaConsumer.java +++ b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java @@ -1,4 +1,4 @@ -package com.baeldung.apachekafka3.groupId; +package com.baeldung.spring.kafka.groupId; import java.util.concurrent.CountDownLatch; diff --git a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaProducer.java b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java similarity index 94% rename from apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaProducer.java rename to spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java index 8707ce800d..f8642a99a5 100644 --- a/apache-kafka-3/src/main/java/com/baeldung/apachekafka3/groupId/MyKafkaProducer.java +++ b/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java @@ -1,4 +1,4 @@ -package com.baeldung.apachekafka3.groupId; +package com.baeldung.spring.kafka.groupId; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/apache-kafka-3/src/test/java/com/baeldung/apachekafka3/groupId/MainLiveTest.java b/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java similarity index 92% rename from apache-kafka-3/src/test/java/com/baeldung/apachekafka3/groupId/MainLiveTest.java rename to spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java index e10e41838a..fb767f1f10 100644 --- a/apache-kafka-3/src/test/java/com/baeldung/apachekafka3/groupId/MainLiveTest.java +++ b/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java @@ -1,4 +1,4 @@ -package com.baeldung.apachekafka3.groupId; +package com.baeldung.spring.kafka.groupId; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsString; @@ -15,7 +15,7 @@ import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.test.annotation.DirtiesContext; @SpringBootTest(classes = Main.class) -@ComponentScan(basePackages = "com.baeldung.apachekafka3.groupId") +@ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") @DirtiesContext @EmbeddedKafka(partitions = 4, topics = { "${kafka.topic.name:test-topic}" }, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" }) public class MainLiveTest { From e7be980d347458d2a13c98fc999eeccac73359ce Mon Sep 17 00:00:00 2001 From: s9m33r Date: Fri, 2 Feb 2024 01:23:09 +0530 Subject: [PATCH 5/8] BAEL-7351 running Embedded Kafka on a separate port for groupId tests --- .../java/com/baeldung/spring/kafka/groupId/MainLiveTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java b/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java index fb767f1f10..dad1dbbf5d 100644 --- a/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java +++ b/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java @@ -17,7 +17,7 @@ import org.springframework.test.annotation.DirtiesContext; @SpringBootTest(classes = Main.class) @ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") @DirtiesContext -@EmbeddedKafka(partitions = 4, topics = { "${kafka.topic.name:test-topic}" }, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" }) +@EmbeddedKafka(partitions = 4, topics = { "${kafka.topic.name:test-topic}" }, brokerProperties = { "listeners=PLAINTEXT://localhost:8000", "port=8000" }) public class MainLiveTest { @Autowired From 95379f73cc089105c9e097986a1af4c52c1d75d3 Mon Sep 17 00:00:00 2001 From: s9m33r Date: Sun, 4 Feb 2024 11:38:52 +0530 Subject: [PATCH 6/8] BAEL-7351 removed new lines --- .../java/com/baeldung/spring/kafka/groupId/MainLiveTest.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java b/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java index dad1dbbf5d..338ed2855f 100644 --- a/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java +++ b/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java @@ -33,9 +33,7 @@ public class MainLiveTest { @Test public void givenEmbeddedKafkaBroker_whenSendingWithSimpleProducer_thenMessageReceived() throws Exception { String data = "Test 123..."; - producer.send(data); - boolean messageConsumed = consumer.getLatch() .await(10, TimeUnit.SECONDS); assertTrue(messageConsumed); From 45a8c4d66f95b1c303a83b8030d7d27d22a4d680 Mon Sep 17 00:00:00 2001 From: s9m33r Date: Sun, 25 Feb 2024 14:55:06 +0530 Subject: [PATCH 7/8] revert unnecessary module re-ordering --- pom.xml | 2 +- .../java/com/baeldung/spring/kafka/groupId/KafkaConfig.java | 0 .../com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java | 0 .../src/main/java/com/baeldung/spring/kafka/groupId/Main.java | 0 .../java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java | 0 .../java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java | 0 .../java/com/baeldung/spring/kafka/groupId/MainLiveTest.java | 0 7 files changed, 1 insertion(+), 1 deletion(-) rename {spring-kafka-2 => spring-kafka-3}/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java (100%) rename {spring-kafka-2 => spring-kafka-3}/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java (100%) rename {spring-kafka-2 => spring-kafka-3}/src/main/java/com/baeldung/spring/kafka/groupId/Main.java (100%) rename {spring-kafka-2 => spring-kafka-3}/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java (100%) rename {spring-kafka-2 => spring-kafka-3}/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java (100%) rename {spring-kafka-2 => spring-kafka-3}/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java (100%) diff --git a/pom.xml b/pom.xml index 330035020f..030b8e1843 100644 --- a/pom.xml +++ b/pom.xml @@ -670,8 +670,8 @@ apache-httpclient-2 apache-httpclient4 apache-httpclient - apache-kafka apache-kafka-2 + apache-kafka apache-libraries-2 apache-libraries apache-olingo diff --git a/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java similarity index 100% rename from spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java rename to spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaConfig.java diff --git a/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java similarity index 100% rename from spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java rename to spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/KafkaErrorHandler.java diff --git a/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/Main.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java similarity index 100% rename from spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/Main.java rename to spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/Main.java diff --git a/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java similarity index 100% rename from spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java rename to spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaConsumer.java diff --git a/spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java b/spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java similarity index 100% rename from spring-kafka-2/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java rename to spring-kafka-3/src/main/java/com/baeldung/spring/kafka/groupId/MyKafkaProducer.java diff --git a/spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java b/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java similarity index 100% rename from spring-kafka-2/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java rename to spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java From 7ade810d7697a53d9602f1b58dd06d0ae92f0fd0 Mon Sep 17 00:00:00 2001 From: s9m33r Date: Sun, 25 Feb 2024 15:27:47 +0530 Subject: [PATCH 8/8] BAEL-7351 test configuration --- .../java/com/baeldung/spring/kafka/groupId/MainLiveTest.java | 2 ++ .../src/test/resources/application-groupId.properties | 1 + 2 files changed, 3 insertions(+) create mode 100644 spring-kafka-3/src/test/resources/application-groupId.properties diff --git a/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java b/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java index 338ed2855f..2c48d25ca2 100644 --- a/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java +++ b/spring-kafka-3/src/test/java/com/baeldung/spring/kafka/groupId/MainLiveTest.java @@ -13,8 +13,10 @@ import org.springframework.boot.test.context.SpringBootTest; import org.springframework.context.annotation.ComponentScan; import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ActiveProfiles; @SpringBootTest(classes = Main.class) +@ActiveProfiles("groupId") @ComponentScan(basePackages = "com.baeldung.spring.kafka.groupId") @DirtiesContext @EmbeddedKafka(partitions = 4, topics = { "${kafka.topic.name:test-topic}" }, brokerProperties = { "listeners=PLAINTEXT://localhost:8000", "port=8000" }) diff --git a/spring-kafka-3/src/test/resources/application-groupId.properties b/spring-kafka-3/src/test/resources/application-groupId.properties new file mode 100644 index 0000000000..648ae1b603 --- /dev/null +++ b/spring-kafka-3/src/test/resources/application-groupId.properties @@ -0,0 +1 @@ +spring.kafka.bootstrap-servers=localhost:8000