Feat (Core): PostgresJobQueue Implementation and Job Processing Enhancements

[jgambarios]

This PR introduces the PostgresJobQueue implementation, fulfilling the requirement for a default job queue using PostgreSQL.
In the process of developing this implementation, we've made significant improvements to the overall job processing system, including enhancements to the JobQueueManagerAPIImpl and related components.

Key Features of PostgresJobQueue:

1. Efficient Job Management: Implements all required operations for job queue management, including creating, retrieving, updating, and monitoring jobs.
2. JSONB Field Usage: Utilizes PostgreSQL's JSONB fields for storing job parameters and results, allowing for flexible and efficient data storage.
3. Optimized Queries: Implements efficient SQL queries, including the use of SELECT FOR UPDATE SKIP LOCKED for concurrent job processing.
4. Pagination Support: Provides paginated results for job listings, improving performance for large job sets.
5. Comprehensive Job Lifecycle Management: Handles all stages of a job's lifecycle, from creation to completion or failure.
6. Transactional Integrity: Ensures data consistency through proper transaction management.

Improvements to Job Processing System:

1. Enhanced Error Handling: Implemented a more robust error handling mechanism, including the use of custom exceptions for different error scenarios.
2. Improved Retry Logic: Refined the retry mechanism to handle transient failures more gracefully.
3. Real-Time Job Monitoring: Enhanced the real-time monitoring capabilities, allowing for more responsive job status updates.
4. Progress Tracking: Improved the progress tracking system, providing more accurate and timely updates on job completion percentage.
5. Cancellation Support: Implemented a more reliable job cancellation process.

JobQueueManagerAPIImpl Enhancements:

1. Event-Driven Updates: Implemented an event-driven system for job status updates, improving system responsiveness.
2. More Granular Job State Transitions: Implemented more detailed job state transitions, providing better insights into job processing stages.

Testing and Documentation:

Developed an extensive set of unit and integration tests for the PostgresJobQueue implementation.

Simple class diagram with main operations and interactions

classDiagram
    class JobQueueManagerAPI {
        <<interface>>
        +start()
        +close()
        +createJob(queueName: String, parameters: Map): String
        +getJob(jobId: String): Job
        +getJobs(page: int, pageSize: int): JobPaginatedResult
        +cancelJob(jobId: String)
        +watchJob(jobId: String, watcher: Consumer~Job~)
        +setRetryStrategy(queueName: String, retryStrategy: RetryStrategy)
    }

    class JobQueueManagerAPIImpl {
        -jobQueue: JobQueue
        -processors: Map~String, JobProcessor~
        -retryStrategies: Map~String, RetryStrategy~
        -circuitBreaker: CircuitBreaker
        +start()
        +close()
        -processJobs()
        -processJobWithRetry(job: Job)
    }

    class JobQueue {
        <<interface>>
        +createJob(queueName: String, parameters: Map): String
        +getJob(jobId: String): Job
        +getActiveJobs(queueName: String, page: int, pageSize: int): JobPaginatedResult
        +getCompletedJobs(queueName: String, startDate: LocalDateTime, endDate: LocalDateTime, page: int, pageSize: int): JobPaginatedResult
        +updateJobStatus(job: Job)
        +nextJob(): Job
        +updateJobProgress(jobId: String, progress: float)
    }

    class PostgresJobQueue {
        -objectMapper: ObjectMapper
        +createJob(queueName: String, parameters: Map): String
        +getJob(jobId: String): Job
        +getActiveJobs(queueName: String, page: int, pageSize: int): JobPaginatedResult
        +getCompletedJobs(queueName: String, startDate: LocalDateTime, endDate: LocalDateTime, page: int, pageSize: int): JobPaginatedResult
        +updateJobStatus(job: Job)
        +nextJob(): Job
        +updateJobProgress(jobId: String, progress: float)
    }

    class Job {
        +id: String
        +queueName: String
        +state: JobState
        +parameters: Map
        +progress: float
        +createdAt: LocalDateTime
        +updatedAt: LocalDateTime
        +startedAt: Optional~LocalDateTime~
        +completedAt: Optional~LocalDateTime~
        +result: Optional~JobResult~
    }

    class JobProcessor {
        <<interface>>
        +process(job: Job)
        +getResultMetadata(job: Job): Map
    }

    class Cancellable {
        <<interface>>
        +cancel(job: Job)
    }

    class RetryStrategy {
        <<interface>>
        +shouldRetry(job: Job, exceptionClass: Class): boolean
        +nextRetryDelay(job: Job): long
    }

    class CircuitBreaker {
        -failureThreshold: int
        -resetTimeout: long
        +allowRequest(): boolean
        +recordFailure()
        +reset()
    }

    class RealTimeJobMonitor {
        -jobWatchers: Map~String, List~Consumer~Job~~~
        +registerWatcher(jobId: String, watcher: Consumer~Job~)
        +updateWatchers(updatedJobs: List~Job~)
    }

    JobQueueManagerAPI <|.. JobQueueManagerAPIImpl
    JobQueueManagerAPIImpl --> JobQueue
    JobQueueManagerAPIImpl --> JobProcessor
    JobQueueManagerAPIImpl --> RetryStrategy
    JobQueueManagerAPIImpl --> CircuitBreaker
    JobQueueManagerAPIImpl --> RealTimeJobMonitor
    JobQueue <|.. PostgresJobQueue
    JobQueueManagerAPIImpl ..> Job
    JobProcessor <|-- Cancellable

Loading

This PR fixes: #29479

[dotcms-sonarqube]

Quality Gate passed

Issues
0 New issues
2 Fixed issues
2 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube