WFLY-19706 More gracefully handle the job execution status during a server crash.

[liweinan]

No description provided.

[liweinan]

@jamezp I'm working on this direction by first adding a ForceStopJobOperatorImpl class that supports the forceStop() method and then adding a findAllTimoutJobExecutions() method.

These methods can be used to filter out suspicious crashed jobs and provide a method to force-stop them.

For the force-stop process, the batch status in these tables need to be updated:

batch_db=# \dt
                 List of relations
 Schema |        Name         | Type  |   Owner
--------+---------------------+-------+------------
 public | job_execution       | table | batch_user
 public | job_instance        | table | batch_user
 public | partition_execution | table | batch_user
 public | step_execution      | table | batch_user
(4 rows)

[liweinan]

Maybe we can set the exitstatus into CRASHED.

And we need to update the endtime to the current time.

We also need to update the other tables here:

• partition_execution
• step_execution

[jamezp]

This isn't thread safe. You'd really need all this to happen in something similar to a transaction which either all or none are updated.

[liweinan]

Yes, I should put them into a single tx like the implementation in the current stop() method in JdbcRepository :

   @Override
    public void stopJobExecution(final JobExecutionImpl jobExecution) {
        super.stopJobExecution(jobExecution);
        final String[] stopExecutionSqls = {
                sqls.getProperty(STOP_JOB_EXECUTION),
                sqls.getProperty(STOP_STEP_EXECUTION),
                sqls.getProperty(STOP_PARTITION_EXECUTION)
        };
        final String jobExecutionIdString = String.valueOf(jobExecution.getExecutionId());
        final String newBatchStatus = BatchStatus.STOPPING.toString();
        final Connection connection = getConnection();
        Statement stmt = null;
        try {
            stmt = connection.createStatement();
            for (String sql : stopExecutionSqls) {
                stmt.addBatch(sql.replace("?", jobExecutionIdString));
            }
            stmt.executeBatch();
        } catch (Exception e) {
            throw BatchMessages.MESSAGES.failToRunQuery(e, Arrays.toString(stopExecutionSqls));
        } finally {
            close(connection, stmt, null, null);
        }
    }

[liweinan]

Maybe I can move the InfinispanRepostiory and the MongoRepository out of jberet-core to reduce the dependencies. That's another task though.

[jamezp]

Is the idea that you'd query jobs that are in say a STOPPING state, then execute forceStop() on them to change the state?

[liweinan]

Is the idea that you'd query jobs that are in say a STOPPING state, then execute forceStop() on them to change the state?

I plan to query the STOPPING and the STARTING states in the findAllTimoutJobExecutions() method. We can change the logic of this method according to the detail requirement in the future.

And maybe it also needs a query for a single job execution to see if its timeout or not. I'll add it too.

[liweinan]

There are several ways to use these APIs after they are implemented:

• Add a batchlet that uses these new APIs to batch-close the timeout jobs.

In addition, the WildFly Admin CLI and GUI can also use these APIs to provide functions like:

• Find out the timeout jobs.
• Provide force-close() operation on one single or multiple job executions.

I'll work on the above task discuss it on the WildFly side, and write the RFE doc according to the discussion result.

That's my work plan on this topic. I plan to complete this PR as the first step by the end of this year. @jamezp Thanks for reviewing this PR! It's an experimental feature and we can change the implementation if it doesn't work. Please let me know if you have any more suggestions and opinions.

[jamezp]

My personal opinion is forceStop() doesn't seem like the right name to me. It's not really stopping anything, it's just changing the job status. I think it might make more sense to have something named changeJobStatus(BatchStatus expectedStatus, BatchStatus newStatus).

My understanding is the issue attempting to be solved is that a job is in an invalid state and we want to update the state. Nothing is really being stopped as much as the state is being updated.

[liweinan]

@jamezp Thanks for the review! I'll rename the method name. In this method, there are three tables changed accordingly:

• job_execution
• partition_execution
• step_execution

Because they all have the batchstatus and exitstatus needs to be updated:

batch_db=# \d job_execution
                                               Table "public.job_execution"
     Column      |           Type           | Collation | Nullable |                        Default
-----------------+--------------------------+-----------+----------+-------------------------------------------------------
 jobexecutionid  | bigint                   |           | not null | nextval('job_execution_jobexecutionid_seq'::regclass)
 jobinstanceid   | bigint                   |           | not null |
 version         | integer                  |           |          |
 createtime      | timestamp with time zone |           |          |
 starttime       | timestamp with time zone |           |          |
 endtime         | timestamp with time zone |           |          |
 lastupdatedtime | timestamp with time zone |           |          |
 batchstatus     | character varying(30)    |           |          |
 exitstatus      | character varying(512)   |           |          |
 jobparameters   | character varying(3000)  |           |          |
 restartposition | character varying(255)   |           |          |
Indexes:
    "job_execution_pkey" PRIMARY KEY, btree (jobexecutionid)
Foreign-key constraints:
    "fk_job_execution_job_instance" FOREIGN KEY (jobinstanceid) REFERENCES job_instance(jobinstanceid) ON DELETE CASCADE
Referenced by:
    TABLE "step_execution" CONSTRAINT "fk_step_exe_job_exe" FOREIGN KEY (jobexecutionid) REFERENCES job_execution(jobexecutionid) ON DELETE CASCADE

batch_db=# \d step_execution
                                                  Table "public.step_execution"
        Column        |           Type           | Collation | Nullable |                         Default
----------------------+--------------------------+-----------+----------+---------------------------------------------------------
 stepexecutionid      | bigint                   |           | not null | nextval('step_execution_stepexecutionid_seq'::regclass)
 jobexecutionid       | bigint                   |           | not null |
 version              | integer                  |           |          |
 stepname             | character varying(255)   |           |          |
 starttime            | timestamp with time zone |           |          |
 endtime              | timestamp with time zone |           |          |
 batchstatus          | character varying(30)    |           |          |
 exitstatus           | character varying(512)   |           |          |
 executionexception   | character varying(2048)  |           |          |
 persistentuserdata   | bytea                    |           |          |
 readcount            | integer                  |           |          |
 writecount           | integer                  |           |          |
 commitcount          | integer                  |           |          |
 rollbackcount        | integer                  |           |          |
 readskipcount        | integer                  |           |          |
 processskipcount     | integer                  |           |          |
 filtercount          | integer                  |           |          |
 writeskipcount       | integer                  |           |          |
 readercheckpointinfo | bytea                    |           |          |
 writercheckpointinfo | bytea                    |           |          |
Indexes:
    "step_execution_pkey" PRIMARY KEY, btree (stepexecutionid)
Foreign-key constraints:
    "fk_step_exe_job_exe" FOREIGN KEY (jobexecutionid) REFERENCES job_execution(jobexecutionid) ON DELETE CASCADE
Referenced by:
    TABLE "partition_execution" CONSTRAINT "fk_partition_exe_step_exe" FOREIGN KEY (stepexecutionid) REFERENCES step_execution(stepexecutionid) ON DELETE CASCADE

batch_db=# \d partition_execution
                       Table "public.partition_execution"
        Column        |          Type           | Collation | Nullable | Default
----------------------+-------------------------+-----------+----------+---------
 partitionexecutionid | integer                 |           | not null |
 stepexecutionid      | bigint                  |           | not null |
 version              | integer                 |           |          |
 batchstatus          | character varying(30)   |           |          |
 exitstatus           | character varying(512)  |           |          |
 executionexception   | character varying(2048) |           |          |
 persistentuserdata   | bytea                   |           |          |
 readercheckpointinfo | bytea                   |           |          |
 writercheckpointinfo | bytea                   |           |          |
Indexes:
    "partition_execution_pkey" PRIMARY KEY, btree (partitionexecutionid, stepexecutionid)
Foreign-key constraints:
    "fk_partition_exe_step_exe" FOREIGN KEY (stepexecutionid) REFERENCES step_execution(stepexecutionid) ON DELETE CASCADE

batch_db=#

I plan to put these update actions into one transaction.

In the future, if we do need to have a forceStop() method that not only change the statuses of these columns but also take other actions in a single transaction, we can add it then.