Implement stateful jobs in the job queue infrastructure

Objective

Add support for stateful jobs in our job queue system, allowing jobs to be marked as "stateful" to ensure only one job of the same type (queue name) runs at a time.

Background

In our migration from Quartz to our custom job queue system, we need to support the concept of "stateful jobs" where only one job of a specific type can run at a time. This is similar to Quartz's stateful job behavior, which prevents concurrent execution of jobs with the same identity.

Implementation Details

1. Database Schema Updates

Add a stateful flag to the job_queue table:

-- Add stateful column to job_queue table
ALTER TABLE job_queue ADD COLUMN stateful BOOLEAN DEFAULT FALSE;

-- Create index for efficient querying of stateful jobs
CREATE INDEX idx_job_queue_queue_name_stateful ON job_queue (queue_name, stateful);

2. Create Stateful Annotation

Create a marker annotation to identify stateful job processors:

/**
 * Marker annotation to indicate a job processor is stateful.
 * Only one job with the same queue name marked as stateful
 * will be allowed to execute at a time.
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface Stateful {
    // No methods needed - this is a marker annotation
}

3. Update JobQueue Interface

Extend the JobQueue interface to support creating stateful jobs:

/**
 * Creates a new job in the queue with the stateful flag.
 *
 * @param queueName The name of the queue
 * @param parameters The job parameters
 * @param stateful Whether the job is stateful
 * @return The ID of the created job
 * @throws JobQueueException if there's an error creating the job
 */
String createJob(String queueName, Map<String, Object> parameters, boolean stateful)
    throws JobQueueException;

/**
 * Default implementation for backward compatibility.
 */
default String createJob(String queueName, Map<String, Object> parameters)
    throws JobQueueException {
    return createJob(queueName, parameters, false);
}

4. Update PostgresJobQueue Implementation

4.1. Modify createJob method:

@Override
@WrapInTransaction
public String createJob(final String queueName, final Map<String, Object> parameters, final boolean stateful)
        throws JobQueueException {
    // Existing implementation...

    // Update the CREATE_JOB_QUEUE_QUERY SQL to include the stateful parameter
    final static String CREATE_JOB_QUEUE_QUERY =
        "INSERT INTO job_queue (id, queue_name, state, created_at, stateful) VALUES (?, ?, ?, ?, ?)";

    // Add stateful parameter to query execution
    new DotConnect().setSQL(CREATE_JOB_QUEUE_QUERY)
            .addParam(jobId)
            .addParam(queueName)
            .addParam(jobState)
            .addParam(now)
            .addParam(stateful) // Add the stateful flag
            .loadResult();

    // Rest of implementation...
}

4.2. Update nextJob method:

@Override
@CloseDBIfOpened
public Job nextJob() throws JobQueueDataException, JobLockingException {
    try {
        // Modified query to respect stateful job constraints
        String query = "WITH stateful_running AS (" +
                "SELECT queue_name FROM job_queue WHERE stateful = true AND state = ? " +
                ") " +
                "UPDATE job_queue SET state = ? " +
                "WHERE id = (" +
                "SELECT jq.id FROM job_queue jq " +
                "LEFT JOIN stateful_running sr ON jq.queue_name = sr.queue_name " +
                "WHERE jq.state = ? " +
                "AND (sr.queue_name IS NULL OR jq.stateful = false) " +
                "ORDER BY jq.priority DESC, jq.created_at ASC LIMIT 1 " +
                "FOR UPDATE SKIP LOCKED) " +
                "RETURNING *";

        DotConnect dc = new DotConnect();
        dc.setSQL(query);
        dc.addParam(JobState.RUNNING.name());
        dc.addParam(JobState.RUNNING.name());
        dc.addParam(JobState.PENDING.name());

        List<Map<String, Object>> results = dc.loadObjectResults();
        if (!results.isEmpty()) {
            String jobId = (String) results.get(0).get("id");
            return getJob(jobId);
        }

        return null;
    } catch (DotDataException e) {
        Logger.error(this, "Database error while fetching next job", e);
        throw new JobQueueDataException("Database error while fetching next job", e);
    } catch (Exception e) {
        Logger.error(this, "Error while locking next job", e);
        throw new JobLockingException("Error while locking next job: " + e.getMessage());
    }
}

5. Update JobQueueManagerAPI Implementation

@Override
public String createJob(final String queueName, final Map<String, Object> parameters)
        throws JobProcessorNotFoundException, DotDataException {

    // Check if processor is stateful
    final Class<? extends JobProcessor> clazz = processors.get(queueName);
    if (null == clazz) {
        throw new JobProcessorNotFoundException(queueName);
    }

    boolean isStateful = clazz.isAnnotationPresent(Stateful.class);

    try {
        return jobQueue.createJob(queueName, parameters, isStateful);
    } catch (JobQueueException e) {
        throw new DotDataException("Error creating job", e);
    }
}

Work Items

1. Create upgrade task for adding stateful flag to job_queue table

2. Implement Stateful marker annotation

3. Update JobQueue interface to support stateful flag

4. Modify PostgresJobQueue implementation:

   • Update createJob method to store stateful flag
   • Enhance nextJob method to enforce stateful constraints

5. Update JobQueueManagerAPI implementation to detect stateful jobs

6. Write unit and integration tests for stateful job behavior

Expected Behavior

1. Jobs marked with the @Stateful annotation will be stored with stateful=true in the database
2. If a stateful job of a specific queue name is running, no other stateful job of the same queue name will be started until the running job completes
3. Non-stateful jobs can run concurrently with stateful jobs of the same queue name
4. The underlying locking mechanism will still use "FOR UPDATE SKIP LOCKED" to ensure efficient job selection in concurrent environments

Technical Considerations

1. The implementation leverages PostgreSQL's transaction capabilities for coordination
2. The approach maintains backward compatibility with existing code
3. The change is minimally invasive to the current architecture
4. Performance impact should be negligible as we're using efficient SQL queries and indexes

Proposed Objective

Core Features

Proposed Priority

Priority 2 - Important

Acceptance Criteria

1. Developers can mark job processors as stateful using the @Stateful annotation
2. The system prevents concurrent execution of stateful jobs of the same queue name
3. Non-stateful jobs are unaffected by this change and maintain current behavior
4. All existing unit and integration tests continue to pass
5. New tests specifically for stateful job behavior pass

Start by tracing the JobQueue interface, PostgresJobQueue implementation, and JobQueueManagerAPI, then inspect the job_queue schema and existing queue behavior. Add the Stateful annotation, persistence and selection constraints, and unit and integration coverage so stateful jobs do not run concurrently while non-stateful jobs remain unaffected.