Fixed Kafka bug & added a controller to send message using API

Author: quantdevv

Diff for: polling-publisher/polling-service/src/main/java/com/iluwatar/polling/DataRepository.java

@@ -27,18 +27,21 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import org.springframework.stereotype.Repository;
 import javax.annotation.PostConstruct;
+import org.springframework.stereotype.Repository;
 
 
 /**
  * Data repository to keep or store data.
  */
 @Repository
-public final class DataRepository {
+public class DataRepository {
 
   private final Map<Integer, String> dataStorage = new HashMap<>();
 
+  /**
+   * init after map creation ... to put dummy data.
+   */
   @PostConstruct
   public void init() {
     // Injecting dummy data at startup

Diff for: polling-publisher/polling-service/src/main/java/com/iluwatar/polling/DataSourceService.java

@@ -38,6 +38,9 @@ public class DataSourceService {
 
   private final DataRepository repository;
 
+  /**
+   * Constructor & Scheduler to push random data.
+   */
   public DataSourceService(DataRepository repository) {
     this.repository = repository;
 

Diff for: polling-publisher/polling-service/src/main/java/com/iluwatar/polling/KafkaProducer.java

@@ -27,15 +27,16 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.stereotype.Service;
+import org.springframework.stereotype.Component;
 
 
 /**
  * This class is responsible for sending messages to Kafka.
  */
-@Service
+@Component
 public class KafkaProducer {
 
+  @Autowired
   private final KafkaTemplate<String, String> kafkaTemplate;
 
   public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
@@ -47,7 +48,7 @@ public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
    *
    * @param message The message to send.
    */
-  public void sendMessage(String message) {
-    kafkaTemplate.send("updates", message);
+  public void sendMessage(String topic, String message) {
+    kafkaTemplate.send(topic, message);
   }
 }

Diff for: polling-publisher/polling-service/src/main/java/com/iluwatar/polling/PollingController.java

@@ -25,7 +25,10 @@
 
 package com.iluwatar.polling;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
 /**
@@ -38,4 +41,14 @@ public class PollingController {
   public String healthCheck() {
     return "Polling Service is up and running!";
   }
+
+
+  @Autowired
+  private KafkaProducer kafkaProducer;
+
+  @PostMapping("/send")
+  public String sendMessage(@RequestParam("message") String message) {
+    kafkaProducer.sendMessage("api-message", message);
+    return "Message sent: " + message;
+  }
 }

Diff for: polling-publisher/polling-service/src/main/java/com/iluwatar/polling/PollingScheduler.java

@@ -25,10 +25,10 @@
 
 package com.iluwatar.polling;
 
+import java.util.Random;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
-import java.util.Random;
 
 /**
  * This class is responsible for scheduling polling tasks.
@@ -52,7 +52,7 @@ public void pollDataSource() {
 
     if (data != null) {
       System.out.println("🟢 Publishing Data: " + data);
-      kafkaProducer.sendMessage(data);
+      kafkaProducer.sendMessage("update", data);
     } else {
       System.out.println("🔴 No Data Found for ID: " + id);
     }

Diff for: polling-publisher/polling-service/src/main/resources/application.yml

@@ -23,7 +23,12 @@
 
 server:
     port: 8081
+
 spring:
     kafka:
+        bootstrap-servers: localhost:9092 # Kafka broker address(es)
         consumer:
-            group-id: subscriber-group
+            group-id: subscriber-group # Consumer group ID
+            auto-offset-reset: earliest # How to reset offset on consumer startup
+            # Add other consumer properties here
+

Diff for: polling-publisher/polling-service/src/test/java/com/iluwatar/polling/AppTest.java

@@ -24,13 +24,14 @@
  */
 package com.iluwatar.polling;
 
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.test.context.SpringBootTest;
 
 import static org.junit.jupiter.api.Assertions.*;
 
 @SpringBootTest
-class AppTest {
+public class AppTest {
 
   @Test
   void polling() {