diff --git a/kafka-sample/spring-boot-multiple-producers-consumers/pom.xml b/kafka-sample/spring-boot-multiple-producers-consumers/pom.xml index 0570b59d..629197d1 100644 --- a/kafka-sample/spring-boot-multiple-producers-consumers/pom.xml +++ b/kafka-sample/spring-boot-multiple-producers-consumers/pom.xml @@ -104,7 +104,7 @@ - 2.38.0 + 2.30.0 diff --git a/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/JsonReceiver.java b/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/JsonReceiver.java index dd18468a..94d161f4 100644 --- a/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/JsonReceiver.java +++ b/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/JsonReceiver.java @@ -5,14 +5,12 @@ import com.example.springbootkafka.multi.domain.SimpleMessage; import jakarta.validation.Valid; import java.util.concurrent.CountDownLatch; -import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.stereotype.Component; import org.springframework.validation.annotation.Validated; -@Getter @Component @Slf4j @Validated @@ -20,8 +18,12 @@ public class JsonReceiver { private final CountDownLatch latch = new CountDownLatch(1); + public CountDownLatch getLatch() { + return latch; + } + @KafkaListener(topics = TOPIC_TEST_2, containerFactory = "jsonKafkaListenerContainerFactory") - public void jsonListener(@Payload @Valid SimpleMessage cr) { + public void listen2(@Payload @Valid SimpleMessage cr) { log.info(TOPIC_TEST_2 + " Received a message with key=" + cr.key() + ", value=" + cr.value()); latch.countDown(); } diff --git a/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/SimpleReceiver.java b/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/SimpleReceiver.java index 307ab817..34817bff 100644 --- a/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/SimpleReceiver.java +++ b/kafka-sample/spring-boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/receiver/SimpleReceiver.java @@ -3,22 +3,26 @@ import static com.example.springbootkafka.multi.util.AppConstants.TOPIC_TEST_1; import java.util.concurrent.CountDownLatch; -import lombok.Getter; -import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; -@Getter -@Slf4j @Component public class SimpleReceiver { + private static final Logger logger = LoggerFactory.getLogger(SimpleReceiver.class); + private final CountDownLatch latch = new CountDownLatch(1); + public CountDownLatch getLatch() { + return latch; + } + @KafkaListener(topics = TOPIC_TEST_1, containerFactory = "simpleKafkaListenerContainerFactory") - public void simpleListener(ConsumerRecord cr) { - log.info(TOPIC_TEST_1 + " Received a message with key=" + cr.key() + ", value=" + cr.value()); + public void listen1(ConsumerRecord cr) { + logger.info(TOPIC_TEST_1 + " Received a message with key=" + cr.key() + ", value=" + cr.value()); latch.countDown(); } } diff --git a/kafka-sample/spring-boot-multiple-producers-consumers/src/test/java/com/example/springbootkafka/multi/TestSpringBootKafkaMultiApplication.java b/kafka-sample/spring-boot-multiple-producers-consumers/src/test/java/com/example/springbootkafka/multi/TestSpringBootKafkaMultiApplication.java index 6590e486..78bf3c44 100644 --- a/kafka-sample/spring-boot-multiple-producers-consumers/src/test/java/com/example/springbootkafka/multi/TestSpringBootKafkaMultiApplication.java +++ b/kafka-sample/spring-boot-multiple-producers-consumers/src/test/java/com/example/springbootkafka/multi/TestSpringBootKafkaMultiApplication.java @@ -13,7 +13,7 @@ public class TestSpringBootKafkaMultiApplication { @Bean @ServiceConnection KafkaContainer kafkaContainer() { - return new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka").withTag("7.5.2")).withKraft(); + return new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka").withTag("7.5.1")).withKraft(); } public static void main(String[] args) {