BUG: kafka event is not consistent.

Sometimes the events are consumed in Task-Service, while other times it is consumed in TimeEntry-Service; whereas it should be consumed by both services!

src/main/java/com/seyed/ali/timeentryservice/event/ProjectEventListener.java

@@ -1,12 +1,12 @@
 package com.seyed.ali.timeentryservice.event;
 
+import com.seyed.ali.timeentryservice.model.enums.OperationType;
 import com.seyed.ali.timeentryservice.model.payload.ProjectDTO;
-import com.seyed.ali.timeentryservice.repository.TimeEntryRepository;
-import com.seyed.ali.timeentryservice.service.interfaces.TimeEntryService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.stereotype.Service;
 
@@ -15,19 +15,19 @@
 @RequiredArgsConstructor
 public class ProjectEventListener {
 
-    private final TimeEntryService timeEntryService;
-    private final TimeEntryRepository timeEntryRepository;
+    private final ProjectEventService projectEventService;
 
     @KafkaListener(
             topics = "${spring.kafka.topic.name}",
             groupId = "${spring.kafka.consumer.group-id}" // if we change the group-id name, the events from the previous data will also be logged, otherwise, only the new events will be logged!
     )
-    public void listenProject(@Payload ConsumerRecord<String, ProjectDTO> record) {
+    public void handleProjectEvent(@Payload ConsumerRecord<String, ProjectDTO> record, @Header("OperationType") String operationType) {
         ProjectDTO projectDTO = record.value();
-        log.info("Received message: {}", projectDTO);
-
-        this.timeEntryRepository.findByProjectId(projectDTO.getProjectId())
-                .forEach(this.timeEntryService::deleteTimeEntry);
+        switch (OperationType.valueOf(operationType)) {
+            case DELETE -> this.projectEventService.handleDeleteOperation(projectDTO);
+            case DETACH -> this.projectEventService.handleDetachOperation(projectDTO);
+            default -> log.warn("\"{}\" operation type not supported.", operationType);
+        }
     }
 
 }

src/main/java/com/seyed/ali/timeentryservice/event/ProjectEventService.java

@@ -0,0 +1,36 @@
+package com.seyed.ali.timeentryservice.event;
+
+import com.seyed.ali.timeentryservice.model.payload.ProjectDTO;
+import com.seyed.ali.timeentryservice.repository.TimeEntryRepository;
+import com.seyed.ali.timeentryservice.service.interfaces.TimeEntryService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.Optional;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ProjectEventService {
+
+    private final TimeEntryService timeEntryService;
+    private final TimeEntryRepository timeEntryRepository;
+
+    public void handleDeleteOperation(ProjectDTO projectDTO) {
+        log.info("\"Delete\" project, associated tasks & time entries: {}", projectDTO);
+        this.timeEntryRepository.findByProjectId(projectDTO.getProjectId())
+                .forEach(this.timeEntryService::deleteTimeEntry);
+    }
+
+    public void handleDetachOperation(ProjectDTO projectDTO) {
+        log.info("\"Detach\" project, associated tasks & time entries: {}", projectDTO);
+        this.timeEntryRepository.findByProjectId(projectDTO.getProjectId())
+                .forEach(timeEntry -> {
+                    timeEntry.setProjectId(null);
+                    timeEntry.setTaskId(null);
+                    this.timeEntryRepository.save(timeEntry);
+                });
+    }
+
+}

src/main/java/com/seyed/ali/timeentryservice/model/enums/OperationType.java

@@ -0,0 +1,7 @@
+package com.seyed.ali.timeentryservice.model.enums;
+
+public enum OperationType {
+
+    DELETE, DETACH
+
+}

src/main/java/com/seyed/ali/timeentryservice/repository/TimeEntryRepository.java

@@ -14,4 +14,6 @@ public interface TimeEntryRepository extends JpaRepository<TimeEntry, String> {
 
     List<TimeEntry> findByProjectId(String projectId);
 
+    List<TimeEntry> findByTaskId(String taskId);
+
 }

src/main/resources/application-dev.yml

@@ -77,7 +77,7 @@ spring:
   kafka:
     consumer:
       bootstrap-servers: localhost:9092
-      group-id: project_group
+      group-id: project_group_timeentry
       auto-offset-reset: earliest
 
       #configure deserialize classes for key & value pair