[ML] Fix submit after shutdown in process worker service (#83646)

Backport of #83645

Author: davidkyle

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectWorkerExecutorService.java

@@ -75,6 +75,7 @@ public synchronized void execute(Runnable command) {
             EsRejectedExecutionException rejected = new EsRejectedExecutionException("autodetect worker service has shutdown", true);
             if (command instanceof AbstractRunnable) {
                 ((AbstractRunnable) command).onRejection(rejected);
+                return;
             } else {
                 throw rejected;
             }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectWorkerExecutorServiceTests.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.isA;
 
 public class AutodetectWorkerExecutorServiceTests extends ESTestCase {
 
@@ -39,6 +40,33 @@ public void testAutodetectWorkerExecutorService_SubmitAfterShutdown() {
         expectThrows(EsRejectedExecutionException.class, () -> executor.execute(() -> {}));
     }
 
+    public void testAutodetectWorkerExecutorService_SubmitAfterShutdownWithAbstractRunnable() {
+        AutodetectWorkerExecutorService executor = new AutodetectWorkerExecutorService(new ThreadContext(Settings.EMPTY));
+
+        threadPool.generic().execute(() -> executor.start());
+        executor.shutdown();
+        AtomicBoolean rejected = new AtomicBoolean(false);
+        executor.execute(new AbstractRunnable() {
+            @Override
+            public void onRejection(Exception e) {
+                assertThat(e, isA(EsRejectedExecutionException.class));
+                rejected.set(true);
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                fail("onFailure should not be called after the worker is shutdown");
+            }
+
+            @Override
+            protected void doRun() throws Exception {
+                fail("doRun should not be called after the worker is shutdown");
+            }
+        });
+
+        assertTrue(rejected.get());
+    }
+
     public void testAutodetectWorkerExecutorService_TasksNotExecutedCallHandlerOnShutdown() throws Exception {
         AutodetectWorkerExecutorService executor = new AutodetectWorkerExecutorService(new ThreadContext(Settings.EMPTY));