[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();
    }

[DonalEvans]

Good idea, this is better

[davidkyle]

Another option is make OrderedRunnable extend AbstractRunnable and implement the required methods as calling the runnable parameter

[DonalEvans]

This solution adds more code than the original one, but I think it might be better, since it simplifies the logic for determining what we need to notify.

[elasticsearchmachine]

💔 Backport failed

Status	Branch	Result
✅	9.2
❌	8.19	Commit could not be cherrypicked due to conflicts
❌	9.1	Commit could not be cherrypicked due to conflicts
❌	9.0	Commit could not be cherrypicked due to conflicts
❌	8.18	Commit could not be cherrypicked due to conflicts

You can use sqren/backport to manually backport by running backport --upstream elastic/elasticsearch --pr 135966

[DonalEvans]

💚 All backports created successfully

Status	Branch	Result
✅	9.1
✅	9.0
✅	8.19
✅	8.18

Questions ?

Please refer to the Backport tool documentation