using postgres native sql FOR UPDATE SKIP LOCKED, to assign workerid to resolve the issue with concurrent instances trying to acquire the jobs

Author: natrajmolala

task-manager-service/src/main/java/uk/gov/hmcts/cp/taskmanager/autoconfig/TaskManagerAutoConfiguration.java

@@ -12,6 +12,8 @@
 
 import jakarta.persistence.EntityManager;
 import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.AutoConfigurationPackages;
@@ -24,13 +26,70 @@
 import org.springframework.core.type.AnnotationMetadata;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.support.JpaRepositoryFactory;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.transaction.PlatformTransactionManager;
 
 @AutoConfiguration
 @Import(TaskManagerAutoConfiguration.PersistencePackagesRegistrar.class)
 @ConditionalOnClass({EntityManager.class, JpaRepository.class, JpaRepositoryFactory.class, PlatformTransactionManager.class})
 public class TaskManagerAutoConfiguration {
 
+    /**
+     * Polling interval in milliseconds for checking unassigned jobs.
+     * Configurable via {@code job.executor.poll-interval} property.
+     */
+    @Value("${job.executor.poll-interval:5000}")
+    private long pollInterval;
+
+    /**
+     * Core thread pool size.
+     * Configurable via {@code job.executor.core-pool-size} property.
+     */
+    @Value("${job.executor.core-pool-size:5}")
+    private int corePoolSize;
+
+    /**
+     * Maximum thread pool size.
+     * Configurable via {@code job.executor.max-pool-size} property.
+     */
+    @Value("${job.executor.max-pool-size:10}")
+    private int maxPoolSize;
+
+    /**
+     * Queue capacity for pending job executions.
+     * Configurable via {@code job.executor.queue-capacity} property.
+     */
+    @Value("${job.executor.queue-capacity:100}")
+    private int queueCapacity;
+
+    /**
+     * Thread name prefix for worker threads.
+     * Configurable via {@code job.executor.thread-name-prefix} property.
+     */
+    @Value("${job.executor.thread-name-prefix:job-executor-}")
+    private String threadNamePrefix;
+
+    /**
+     * Whether to wait for tasks to complete on shutdown.
+     * Configurable via {@code job.executor.wait-for-tasks-on-shutdown} property.
+     */
+    @Value("${job.executor.wait-for-tasks-on-shutdown:true}")
+    private boolean waitForTasksOnShutdown;
+
+    /**
+     * Maximum seconds to wait for task completion on shutdown.
+     * Configurable via {@code job.executor.await-termination-seconds} property.
+     */
+    @Value("${job.executor.await-termination-seconds:60}")
+    private int awaitTerminationSeconds;
+
+    /**
+     * Batch size for fetching unassigned jobs per polling cycle.
+     * Configurable via {@code job.executor.batch-size} property.
+     */
+    @Value("${job.executor.batch-size:50}")
+    private int batchSize;
+
     @Bean(name = "jobsRepository")
     @ConditionalOnMissingBean(name = "jobsRepository")
     public JobsRepository jobsRepository(final EntityManager entityManager) {
@@ -64,9 +123,11 @@ public TaskRegistry taskRegistry(final ObjectProvider<ExecutableTask> executable
     public JobExecutor jobExecutor(
             final JobService jobService,
             final TaskRegistry taskRegistry,
-            final PlatformTransactionManager platformTransactionManager
+            final PlatformTransactionManager platformTransactionManager,
+            @Qualifier("jobExecutorThreadPool")
+            final ThreadPoolTaskExecutor jobExecutorThreadPool
     ) {
-        return new JobExecutor(jobService, taskRegistry, platformTransactionManager);
+        return new JobExecutor(jobService, taskRegistry, platformTransactionManager, jobExecutorThreadPool);
     }
 
     @Bean
@@ -75,6 +136,19 @@ public ExecutionService executionService(final JobService jobService, final Task
         return new ExecutionService(jobService, taskRegistry);
     }
 
+    @Bean(name = "jobExecutorThreadPool")
+    public ThreadPoolTaskExecutor jobExecutorThreadPool() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(corePoolSize);
+        executor.setMaxPoolSize(maxPoolSize);
+        executor.setQueueCapacity(queueCapacity);
+        executor.setThreadNamePrefix(threadNamePrefix);
+        executor.setWaitForTasksToCompleteOnShutdown(waitForTasksOnShutdown);
+        executor.setAwaitTerminationSeconds(awaitTerminationSeconds);
+        executor.initialize();
+        return executor;
+    }
+
     static final class PersistencePackagesRegistrar implements ImportBeanDefinitionRegistrar {
         @Override
         public void registerBeanDefinitions(AnnotationMetadata metadata, BeanDefinitionRegistry registry) {

task-manager-service/src/main/java/uk/gov/hmcts/cp/taskmanager/domain/executor/JobExecutor.java

@@ -1,6 +1,5 @@
 package uk.gov.hmcts.cp.taskmanager.domain.executor;
 
-import org.springframework.transaction.annotation.Transactional;
 import uk.gov.hmcts.cp.taskmanager.service.task.TaskRegistry;
 import uk.gov.hmcts.cp.taskmanager.persistence.entity.Job;
 import uk.gov.hmcts.cp.taskmanager.persistence.service.JobService;
@@ -19,19 +18,19 @@
 
 /**
  * Main scheduler component that polls for unassigned jobs and assigns them to worker threads.
- * 
+ *
  * <p>The JobExecutor is the core component of the task management system. It periodically
  * polls the database for unassigned jobs that are ready to be executed and assigns them
  * to worker threads for processing.
- * 
+ *
  * <p><b>Key responsibilities:</b>
  * <ul>
  *   <li>Periodically polling for unassigned jobs (via {@link Scheduled} annotation)</li>
  *   <li>Assigning jobs to worker threads</li>
  *   <li>Managing a thread pool for concurrent job execution</li>
  *   <li>Handling job assignment failures and retry logic</li>
  * </ul>
- * 
+ *
  * <p><b>Configuration:</b>
  * <p>All thread pool and polling parameters are configurable via application properties:
  * <ul>
@@ -44,7 +43,7 @@
  *   <li>{@code job.executor.await-termination-seconds} - Await termination seconds (default: 60)</li>
  *   <li>{@code job.executor.batch-size} - Batch size for fetching jobs (default: 50)</li>
  * </ul>
- * 
+ *
  * <p><b>Execution flow:</b>
  * <ol>
  *   <li>{@link #checkAndAssignJobs()} is called periodically by Spring's scheduler</li>
@@ -58,10 +57,10 @@
  *   </li>
  *   <li>Job execution is handled by {@link TaskExecutor} in a separate thread</li>
  * </ol>
- * 
+ *
  * <p>The thread pool is initialized on startup ({@link PostConstruct}) and shutdown
  * gracefully on application shutdown ({@link PreDestroy}).
- * 
+ *
  * @author Task Manager Service
  * @since 1.0.0
  * @see TaskExecutor
@@ -79,71 +78,22 @@ public class JobExecutor {
      * Service for job persistence operations.
      */
     private final JobService jobService;
-    
+
     /**
      * Registry for task lookup.
      */
     private final TaskRegistry taskRegistry;
-    
+
     /**
      * Transaction manager for transactional operations.
      */
     private final PlatformTransactionManager transactionManager;
-    
+
     /**
      * Thread pool executor for concurrent job execution.
      */
     private final ThreadPoolTaskExecutor executor;
 
-    /**
-     * Polling interval in milliseconds for checking unassigned jobs.
-     * Configurable via {@code job.executor.poll-interval} property.
-     */
-    @Value("${job.executor.poll-interval:5000}")
-    private long pollInterval;
-
-    /**
-     * Core thread pool size.
-     * Configurable via {@code job.executor.core-pool-size} property.
-     */
-    @Value("${job.executor.core-pool-size:5}")
-    private int corePoolSize;
-
-    /**
-     * Maximum thread pool size.
-     * Configurable via {@code job.executor.max-pool-size} property.
-     */
-    @Value("${job.executor.max-pool-size:10}")
-    private int maxPoolSize;
-
-    /**
-     * Queue capacity for pending job executions.
-     * Configurable via {@code job.executor.queue-capacity} property.
-     */
-    @Value("${job.executor.queue-capacity:100}")
-    private int queueCapacity;
-
-    /**
-     * Thread name prefix for worker threads.
-     * Configurable via {@code job.executor.thread-name-prefix} property.
-     */
-    @Value("${job.executor.thread-name-prefix:job-executor-}")
-    private String threadNamePrefix;
-
-    /**
-     * Whether to wait for tasks to complete on shutdown.
-     * Configurable via {@code job.executor.wait-for-tasks-on-shutdown} property.
-     */
-    @Value("${job.executor.wait-for-tasks-on-shutdown:true}")
-    private boolean waitForTasksOnShutdown;
-
-    /**
-     * Maximum seconds to wait for task completion on shutdown.
-     * Configurable via {@code job.executor.await-termination-seconds} property.
-     */
-    @Value("${job.executor.await-termination-seconds:60}")
-    private int awaitTerminationSeconds;
-
     /**
      * Batch size for fetching unassigned jobs per polling cycle.
      * Configurable via {@code job.executor.batch-size} property.
@@ -153,38 +103,23 @@ public class JobExecutor {
 
     /**
      * Constructs a new JobExecutor with the specified dependencies.
-     * 
+     *
      * @param jobService the job service for persistence operations, must not be null
      * @param taskRegistry the task registry for task lookup, must not be null
      * @param transactionManager the transaction manager, must not be null
      */
-    public JobExecutor(JobService jobService, TaskRegistry taskRegistry, PlatformTransactionManager transactionManager) {
+    public JobExecutor(JobService jobService, TaskRegistry taskRegistry,
+                       PlatformTransactionManager transactionManager,
+                       ThreadPoolTaskExecutor jobExecutorThreadPool) {
         this.jobService = jobService;
         this.taskRegistry = taskRegistry;
         this.transactionManager = transactionManager;
-        this.executor = new ThreadPoolTaskExecutor();
-    }
-
-    /**
-     * Initializes the thread pool executor with configured parameters.
-     * 
-     * <p>This method is called after dependency injection completes. It configures
-     * the thread pool with all the settings from application properties and initializes it.
-     */
-    @PostConstruct
-    public void init() {
-        executor.setCorePoolSize(corePoolSize);
-        executor.setMaxPoolSize(maxPoolSize);
-        executor.setQueueCapacity(queueCapacity);
-        executor.setThreadNamePrefix(threadNamePrefix);
-        executor.setWaitForTasksToCompleteOnShutdown(waitForTasksOnShutdown);
-        executor.setAwaitTerminationSeconds(awaitTerminationSeconds);
-        executor.initialize();
+        this.executor = jobExecutorThreadPool;
     }
 
     /**
      * Shuts down the thread pool executor gracefully.
-     * 
+     *
      * <p>This method is called when the application is shutting down. It ensures
      * that all running tasks complete before the executor is terminated.
      */
@@ -195,7 +130,7 @@ public void destroy() {
 
     /**
      * Periodically checks for unassigned jobs and assigns them to worker threads.
-     * 
+     *
      * <p>This method is scheduled to run at fixed intervals (configurable via
      * {@code job.executor.poll-interval}). It:
      * <ol>
@@ -209,20 +144,19 @@ public void destroy() {
      *     </ul>
      *   </li>
      * </ol>
-     * 
+     *
      * <p>Jobs are ordered by priority (ascending, 1 is highest) and then by start time
      * (ascending), ensuring high-priority jobs are processed first.
-     * 
+     *
      * <p>If an error occurs during job assignment, the retry attempts are decremented
      * so the job can be retried in the next polling cycle.
      */
     @Scheduled(fixedDelayString = "${job.executor.poll-interval:5000}")
-
     public void checkAndAssignJobs() {
         try {
             logger.debug("Checking for unassigned jobs (batch size: {})...", batchSize);
-            UUID workerId = UUID.randomUUID();
-            List<Job> assignedJobs = jobService.assignJobsToWorkerBatch(workerId, ZonedDateTime.now(),ZonedDateTime.now(),batchSize);
+            final UUID workerId = UUID.randomUUID();
+            final List<Job> assignedJobs = jobService.assignJobsToWorkerBatch(workerId, ZonedDateTime.now(), ZonedDateTime.now(), batchSize);
 
             if (assignedJobs.isEmpty()) {
                 logger.debug("No unassigned jobs found");
@@ -252,11 +186,11 @@ public void checkAndAssignJobs() {
 
     /**
      * Executes a job by creating a TaskExecutor and submitting it to the thread pool.
-     * 
+     *
      * <p>This method creates a new {@link TaskExecutor} instance for the job and submits
      * it to the thread pool for execution. The TaskExecutor will handle the actual task
      * execution, including transaction management and error handling.
-     * 
+     *
      * @param job the job to execute, must not be null and must have a worker assigned
      */
     private void executeJob(Job job) {