Feat (Core): PostgresJobQueue Implementation and Job Processing Enhancements

dotCMS/src/main/java/com/dotcms/jobs/business/api/JobQueueConfig.java

@@ -10,13 +10,18 @@ public class JobQueueConfig {
      */
     private final int threadPoolSize;
 
+    // The interval in milliseconds to poll for job updates
+    private final int pollJobUpdatesIntervalMilliseconds;
+
     /**
      * Constructs a new JobQueueConfig
      *
      * @param threadPoolSize The number of threads to use for job processing.
+     * @param pollJobUpdatesIntervalMilliseconds The interval in milliseconds to poll for job updates.
      */
-    public JobQueueConfig(int threadPoolSize) {
+    public JobQueueConfig(int threadPoolSize, int pollJobUpdatesIntervalMilliseconds) {
         this.threadPoolSize = threadPoolSize;
+        this.pollJobUpdatesIntervalMilliseconds = pollJobUpdatesIntervalMilliseconds;
     }
 
     /**
@@ -28,4 +33,13 @@ public int getThreadPoolSize() {
         return threadPoolSize;
     }
 
+    /**
+     * Gets the interval in milliseconds to poll for job updates.
+     *
+     * @return The interval in milliseconds to poll for job updates.
+     */
+    public int getPollJobUpdatesIntervalMilliseconds() {
+        return pollJobUpdatesIntervalMilliseconds;
+    }
+
 }

dotCMS/src/main/java/com/dotcms/jobs/business/api/JobQueueConfigProducer.java

@@ -16,6 +16,11 @@ public class JobQueueConfigProducer {
             "JOB_QUEUE_THREAD_POOL_SIZE", 10
     );
 
+    // The interval in milliseconds to poll for job updates.
+    static final int DEFAULT_POLL_JOB_UPDATES_INTERVAL_MILLISECONDS = Config.getIntProperty(
+            "JOB_QUEUE_POLL_JOB_UPDATES_INTERVAL_MILLISECONDS", 1000
+    );
+
     /**
      * Produces a JobQueueConfig object. This method is called by the CDI container to create a
      * JobQueueConfig instance when it is necessary for dependency injection.
@@ -24,7 +29,10 @@ public class JobQueueConfigProducer {
      */
     @Produces
     public JobQueueConfig produceJobQueueConfig() {
-        return new JobQueueConfig(DEFAULT_THREAD_POOL_SIZE);
+        return new JobQueueConfig(
+                DEFAULT_THREAD_POOL_SIZE,
+                DEFAULT_POLL_JOB_UPDATES_INTERVAL_MILLISECONDS
+        );
     }
 
 }

dotCMS/src/main/java/com/dotcms/jobs/business/api/JobQueueManagerAPI.java

@@ -1,12 +1,13 @@
 package com.dotcms.jobs.business.api;
 
 import com.dotcms.jobs.business.error.CircuitBreaker;
-import com.dotcms.jobs.business.error.JobCancellationException;
-import com.dotcms.jobs.business.error.ProcessorNotFoundException;
+import com.dotcms.jobs.business.error.JobProcessorNotFoundException;
 import com.dotcms.jobs.business.error.RetryStrategy;
 import com.dotcms.jobs.business.job.Job;
+import com.dotcms.jobs.business.job.JobPaginatedResult;
 import com.dotcms.jobs.business.processor.JobProcessor;
-import java.util.List;
+import com.dotcms.jobs.business.queue.JobQueue;
+import com.dotmarketing.exception.DotDataException;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -15,7 +16,7 @@
  * Defines the contract for interacting with the job queue system. This interface provides methods
  * for managing jobs, processors, and the overall state of the job queue.
  */
-public interface JobQueueManagerAPI extends AutoCloseable {
+public interface JobQueueManagerAPI {
 
     /**
      * Starts the job queue manager, initializing the thread pool for job processing.
@@ -62,35 +63,38 @@ public interface JobQueueManagerAPI extends AutoCloseable {
      * @param queueName  The name of the queue
      * @param parameters The parameters for the job
      * @return The ID of the created job
-     * @throws ProcessorNotFoundException if no processor is registered for the specified queue
+     * @throws JobProcessorNotFoundException if no processor is registered for the specified queue
+     * @throws DotDataException              if there's an error creating the job
      */
     String createJob(String queueName, Map<String, Object> parameters)
-            throws ProcessorNotFoundException;
+            throws JobProcessorNotFoundException, DotDataException;
 
     /**
      * Retrieves a job by its ID.
      *
      * @param jobId The ID of the job
      * @return The Job object, or null if not found
+     * @throws DotDataException if there's an error fetching the job
      */
-    Job getJob(String jobId);
+    Job getJob(String jobId) throws DotDataException;
 
     /**
      * Retrieves a list of jobs.
      *
      * @param page     The page number
      * @param pageSize The number of jobs per page
-     * @return A list of Job objects
+     * @return A result object containing the list of active jobs and pagination information.
+     * @throws DotDataException if there's an error fetching the jobs
      */
-    List<Job> getJobs(int page, int pageSize);
+    JobPaginatedResult getJobs(int page, int pageSize) throws DotDataException;
 
     /**
      * Cancels a job.
      *
      * @param jobId The ID of the job to cancel
-     * @throws JobCancellationException if the job cannot be cancelled
+     * @throws DotDataException if there's an error cancelling the job
      */
-    void cancelJob(String jobId) throws JobCancellationException;
+    void cancelJob(String jobId) throws DotDataException;
 
     /**
      * Registers a watcher for a specific job.
@@ -113,6 +117,11 @@ String createJob(String queueName, Map<String, Object> parameters)
      */
     CircuitBreaker getCircuitBreaker();
 
+    /**
+     * @return The JobQueue instance
+     */
+    JobQueue getJobQueue();
+
     /**
      * @return The size of the thread pool
      */