Changes

java 17 -> 21
Update notification scheduler service
- Using PriorityQueue and timer
Added notification sender

Author: LinterProd

Makefile

@@ -2,11 +2,11 @@
 
 build:
 	@echo "🛠️  Билдим backend..."
-	cd task-tracker-backend && mvn clean package -DskipTests
+	mvn clean package -pl task-tracker-backend -DskipTests
 	@echo "🛠️  Билдим email-сервис..."
-	cd task-tracker-email-sender && mvn clean package -DskipTests
+	mvn clean package -pl task-tracker-email-sender -DskipTests
 	@echo "🛠️  Билдим scheduler..."
-	cd task-tracker-scheduler && mvn clean package -DskipTests
+	mvn clean package -pl task-tracker-scheduler -DskipTests
 	@echo "🚀 Запускаем сервисы..."
 	docker-compose up -d --build
 
@@ -20,7 +20,7 @@ stop:
 
 backend:
 	docker-compose down backend
-	cd task-tracker-backend && mvn clean package -DskipTests && cd ..
+	mvn clean package -pl task-tracker-backend -am -DskipTests
 	docker-compose build backend
 
 scheduler:

task-tracker-backend/pom.xml

@@ -27,7 +27,7 @@
         <url/>
     </scm>
     <properties>
-        <java.version>17</java.version>
+        <java.version>21</java.version>
     </properties>
     <dependencies>
         <!-- common -->

task-tracker-backend/src/main/java/com/metarash/backend/service/impl/TaskServiceImpl.java

@@ -44,7 +44,6 @@ public Slice<TaskDto> getTasksByUsername(String username, TaskFilter filter) {
         Specification<Task> spec = TaskSpecifications.build(filter, username);
 
         Slice<TaskDto> tasks = taskRepository.findAll(spec, pageable).map(taskMapper::toDto);
-        log.debug("Found tasks: {}", tasks.getContent());
         return tasks;
     }
 

task-tracker-backend/src/main/java/com/metarash/backend/specification/TaskSpecifications.java

@@ -4,32 +4,35 @@
 import com.metarash.backend.model.entity.Task;
 import org.springframework.data.jpa.domain.Specification;
 
+import java.util.Optional;
+
 public class TaskSpecifications {
 
-    public static Specification<Task> hasUsername(String username) {
+    public static Specification<Task> hasField(String fieldName, Object value) {
         return (root, query, criteriaBuilder) ->
-                criteriaBuilder.equal(root.get("user").get("username"), username);
+                criteriaBuilder.equal(root.get(fieldName), value);
+    }
+
+    public static Specification<Task> hasUsername(String username) {
+        return hasField("username", username);
     }
 
     public static Specification<Task> hasPriority(String priority) {
-        return (root, query, criteriaBuilder) ->
-                criteriaBuilder.equal(root.get("priority"), priority);
+        return hasField("priority", priority);
     }
 
     public static Specification<Task> hasStatus(String status) {
-        return (root, query, criteriaBuilder) ->
-                criteriaBuilder.equal(root.get("status"), status);
+        return hasField("status", status);
     }
 
     public static Specification<Task> build(TaskFilter filter, String username) {
         Specification<Task> spec = Specification.where(hasUsername(username));
 
-        if (filter.getPriority() != null) {
-            spec = spec.and(hasPriority(filter.getPriority()));
-        }
-        if (filter.getStatus() != null) {
-            spec = spec.and(hasStatus(filter.getStatus()));
-        }
+        Optional.ofNullable(filter.getPriority())
+                .ifPresent(priority -> spec.and(hasPriority(priority)));
+
+        Optional.ofNullable(filter.getStatus())
+                .ifPresent(status -> spec.and(hasStatus(status)));
 
         return spec;
     }

task-tracker-email-sender/pom.xml

@@ -27,7 +27,7 @@
         <url/>
     </scm>
     <properties>
-        <java.version>17</java.version>
+        <java.version>21</java.version>
     </properties>
     <dependencies>
         <!-- common -->

task-tracker-scheduler/pom.xml

@@ -16,7 +16,7 @@
     <name>task-tracker-scheduler</name>
     <description>Task Tracker Scheduler</description>
     <properties>
-        <java.version>17</java.version>
+        <java.version>21</java.version>
     </properties>
     <dependencies>
         <!-- common -->

task-tracker-scheduler/pom.xml~

@@ -1,23 +1,26 @@
 package com.metarash.tasktrackerscheduler.kafka.consumer;
 
 import com.metarash.dto.TaskEvent;
-import com.metarash.tasktrackerscheduler.service.OverdueNotificationScheduler;
+import com.metarash.tasktrackerscheduler.service.NotificationScheduler;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
+@Slf4j
 @Component
 @RequiredArgsConstructor
 public class TaskEventListener {
 
-    private final OverdueNotificationScheduler scheduler;
+    private final NotificationScheduler notificationScheduler;
 
     @KafkaListener(
             topics = "task-events",
             groupId = "scheduler-group",
             containerFactory = "kafkaListenerContainerFactory"
     )
     public void onTaskEvent(TaskEvent event) {
-        scheduler.updateDueDateIfEarlier(event.getDueDate());
+        log.info("Received task event: {}", event);
+        notificationScheduler.scheduleNotification(event);
     }
 }

task-tracker-scheduler/src/main/java/com/metarash/tasktrackerscheduler/kafka/consumer/TaskEventListener.java

@@ -0,0 +1,98 @@
+package com.metarash.tasktrackerscheduler.service;
+
+import com.metarash.dto.TaskEvent;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.Comparator;
+import java.util.PriorityQueue;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.locks.ReentrantLock;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class NotificationScheduler {
+    private final NotificationSender notificationSender;
+
+
+    private final PriorityQueue<ScheduledTask> queue =
+            new PriorityQueue<>(Comparator.comparing(ScheduledTask::dueDate));
+
+    private final ReentrantLock lock = new ReentrantLock();
+
+    private final Timer timer = new Timer();
+
+    public void scheduleNotification(TaskEvent event) {
+        lock.lock();
+        try {
+            queue.add(new ScheduledTask(event.getTaskId(), event.getDueDate()));
+            ScheduledTask peek = queue.peek();
+            if (peek == null) {
+                log.info("No tasks in queue - nothing to schedule");
+                return;
+            }
+            if (peek.dueDate().equals(event.getDueDate())) {
+                reschedule();
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void reschedule() {
+        lock.lock();
+        try {
+            timer.cancel();
+
+            ScheduledTask nextTask = queue.peek();
+            if (nextTask == null) {
+                log.info("No tasks in queue - nothing to schedule");
+                return;
+            }
+
+            long delay = Duration.between(LocalDateTime.now(), nextTask.dueDate())
+                    .toMillis();
+
+            if (delay < 0) {
+                log.warn("Task is already overdue: {}", nextTask.taskId());
+                delay = 0;
+            }
+
+            timer.schedule(new TimerTask() {
+                @Override
+                public void run() {
+                    handleDueTask();
+                }
+            }, delay);
+
+            log.info("Scheduled task {} for execution in {} ms", nextTask.taskId(), delay);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void handleDueTask() {
+        lock.lock();
+        try {
+            ScheduledTask task = queue.poll();
+            if (task == null) {
+                log.warn("No task found in queue - possible race condition");
+                return;
+            }
+            notificationSender.sendNotification(task.taskId());
+
+            if (!queue.isEmpty()) {
+                reschedule();
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private record ScheduledTask(Long taskId, LocalDateTime dueDate) {}
+}

task-tracker-scheduler/src/main/java/com/metarash/tasktrackerscheduler/service/NotificationScheduler.java

@@ -0,0 +1,47 @@
+package com.metarash.tasktrackerscheduler.service;
+
+import com.metarash.dto.NotificationMessage;
+import com.metarash.tasktrackerscheduler.entity.Task;
+import com.metarash.tasktrackerscheduler.kafka.producer.ReportProducer;
+import com.metarash.tasktrackerscheduler.repository.TaskRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class NotificationSender {
+    private final ReportProducer reportProducer;
+    private final TaskRepository taskRepository;
+    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
+
+    public void sendNotification(Long taskId) {
+        executor.submit(() -> {
+            try {
+                Task task = taskRepository.findById(taskId)
+                        .orElseThrow(() -> new Exception("task not found with id: " + taskId));
+
+                NotificationMessage message = buildMessage(task);
+                reportProducer.sendOverdueTasks(message);
+                log.info("Sent notification for task {}", taskId);
+            } catch (Exception ex) {
+                log.error("Failed to send notification for task {}", taskId, ex);
+                // Здесь можно добавить повторную попытку
+            }
+        });
+    }
+
+    private NotificationMessage buildMessage(Task task) {
+        return new NotificationMessage(
+                task.getId(),
+                task.getTitle(),
+                String.format("Задача просрочена: '%s'", task.getTitle()),
+                task.getUser().getUsername(),
+                task.getDueDate()
+        );
+    }
+}