Add Kafka dependencies and configuration

Author: Trinh-Tan-Dat

pom.xml

@@ -70,6 +70,16 @@
 			<artifactId>mockito-junit-jupiter</artifactId>
 			<scope>test</scope>
 		</dependency>
+
+		<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>3.2.3</version>
+		</dependency>
+
+
+
 	</dependencies>
 	<build>
 		<plugins>

src/main/java/com/programming/commentService/config/KafkaProducerConfig.java

@@ -0,0 +1,36 @@
+package com.programming.commentService.config;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.core.ProducerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+@Configuration
+public class KafkaProducerConfig {
+
+    @Bean
+    public Map<String, Object> producerConfigs() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        return props;
+    }
+
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        return new DefaultKafkaProducerFactory<>(producerConfigs());
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> kafkaTemplate() {
+        return new KafkaTemplate<>(producerFactory());
+    }
+}

src/main/java/com/programming/commentService/controller/CommentController.java

@@ -5,12 +5,14 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
+import com.google.gson.Gson;
 import com.programming.commentService.model.Comment;
 import com.programming.commentService.repository.CommentRepository;
 import com.programming.commentService.service.CommentService;
@@ -24,6 +26,12 @@
 @RequestMapping("/comment")
 @AllArgsConstructor
 public class CommentController {
+
+    @Autowired
+    private KafkaTemplate<String, String> kafkaTemplate;
+
+    private static final String TOPIC = "new-comments";
+
     private final CommentRepository commentRepository;
 
     @Autowired
@@ -38,7 +46,15 @@ public String getServiceName() {
     @PostMapping("/upload")
     public ResponseEntity<?> uploadComment(@RequestBody Comment comment) {
         try {
+            String commentJson = new Gson().toJson(commentRepository.save(comment));
+            kafkaTemplate.send(TOPIC, commentJson);
+
+
             Comment save = commentRepository.save(comment);
+
+
+            //Produce message to kafka
+
             return ResponseEntity.ok(HttpStatus.CREATED);
         } catch (Exception e) {
             return ResponseEntity.internalServerError().body(e.getMessage());

src/main/resources/application.properties

@@ -0,0 +1,2 @@
+spring.kafka.producer.properties.max.request.size=500000000  # Set the maximum request size for producer
+spring.kafka.producer.properties.buffer.memory=1024000000    # Set the buffer memory for producer

src/main/resources/application.yml

@@ -7,3 +7,14 @@ spring:
   data:
     mongodb:
       uri: ${SPRING_DATA_MONGODB_URI:mongodb://localhost:27017/comment-service}
+      # uri: ${SPRING_DATA_MONGODB_URI:mongodb://localhost:27777/comment-service}
+
+
+  kafka:
+    bootstrap-servers: localhost:9092    # Địa chỉ Kafka broker
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
+    consumer:
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer