[ML] Ensure queued AbstractRunnables are notified when executor stops

[DonalEvans]

AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs.

• Refactor notifyQueueRunnables() to allow PriorityProcessWorkerExecutorService to notify the AbstractRunnable contained within queued OrderedRunnables
• Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start()
• Add unit tests

Closes #134651

[elasticsearchmachine]

Pinging @elastic/ml-core (Team:ML)

[elasticsearchmachine]

Hi @DonalEvans, I've created a changelog YAML for you.

[jonathan-buttner]

Great work, left one suggestion

[jonathan-buttner]

What do you think about having AbstractProcessWorkerExecutorService contain the logic for notifyIfAbstractRunnable and relying on an abstract method like getAsAbstractRunnable which either returns the AbstractRunnable or null. That way child classes don't need to call back into the parent to do the notification, they would only return the abstract runnable if it was one.

Something like:

    protected abstract AbstractRunnable getAsAbstractRunnable(T runnable);

    private void notifyIfAbstractRunnable(T runnable, Exception ex, String msg) {
        var abstractRunnable = getAsAbstractRunnable(runnable);
        if (abstractRunnable != null) {
            notifyAbstractRunnable(ex, msg, abstractRunnable);
        }
    }

Then PriorityProcessWorkerExecutorService would have something like:

    @Override
    protected AbstractRunnable getAsAbstractRunnable(OrderedRunnable orderedRunnable, Exception ex, String msg) {
        // The runnable contained within OrderedRunnable is always an AbstractRunnable, so no need to check the type
        return orderedRunnable.runnable();
    }