Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feat (Core): PostgresJobQueue Implementation and Job Processing Enhan…
…cements (#30175) 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 ```mermaid 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 branch information