framework: do not double-execute an async job after returning its queue item

framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java

@@ -523,7 +523,7 @@ public String  obfuscatePassword(String result, boolean hidePassword) {
         return StringUtils.obfuscatePasswordInJsonLikeString(result);
     }
 
-    private void scheduleExecution(final AsyncJobVO job) {
+    protected void scheduleExecution(final AsyncJobVO job) {
         scheduleExecution(job, false);
     }
 
@@ -701,58 +701,66 @@ private int getAndResetPendingSignals(AsyncJob job) {
         return signals;
     }
 
-    private void executeQueueItem(SyncQueueItemVO item, boolean fromPreviousSession) {
+    /**
+     * Dispatches the job behind a sync queue item.
+     * <p>
+     * TODO: stamping the executing management server id on the job can hit a DB deadlock. As a
+     * temporary workaround, a failed update returns the item to the sync queue so the job is retried
+     * on a later turn instead of failing. The proper fix is to remove the deadlock at its source, in
+     * the locking and transaction around the sync_queue and async_job updates on dispatch, after
+     * which this retry is no longer needed.
+     */
+    protected void executeQueueItem(SyncQueueItemVO item, boolean fromPreviousSession) {
         AsyncJobVO job = _jobDao.findById(item.getContentId());
-        if (job != null) {
+        if (job == null) {
             if (logger.isDebugEnabled()) {
-                logger.debug("Schedule queued job-" + job.getId());
+                logger.debug("Unable to find related job for queue item: " + item.toString());
             }
+            _queueMgr.purgeItem(item.getId());
+            return;
+        }
 
-            job.setSyncSource(item);
-
-            //
-            // TODO: a temporary solution to work-around DB deadlock situation
-            //
-            // to live with DB deadlocks, we will give a chance for job to be rescheduled
-            // in case of exceptions (most-likely DB deadlock exceptions)
-            try {
-                job.setExecutingMsid(getMsid());
-                _jobDao.update(job.getId(), job);
-            } catch (Exception e) {
-                logger.warn("Unexpected exception while dispatching job-" + item.getContentId(), e);
+        if (logger.isDebugEnabled()) {
+            logger.debug("Schedule queued job-" + job.getId());
+        }
+        job.setSyncSource(item);
 
-                try {
-                    _queueMgr.returnItem(item.getId());
-                } catch (Throwable thr) {
-                    logger.error("Unexpected exception while returning job-" + item.getContentId() + " to queue", thr);
-                }
+        try {
+            job.setExecutingMsid(getMsid());
+            _jobDao.update(job.getId(), job);
+        } catch (Exception e) {
+            logger.warn("Unexpected exception while dispatching job-" + item.getContentId(), e);
+            if (returnItemToQueue(item)) {
+                return;
             }
+            // requeue failed: fall through so the job still runs once rather than being stranded until cleanup
+        }
 
-            try {
-                scheduleExecution(job);
-            } catch (RejectedExecutionException e) {
-                logger.warn("Execution for job-" + job.getId() + " is rejected, return it to the queue for next turn");
-
-                try {
-                    _queueMgr.returnItem(item.getId());
-                } catch (Exception e2) {
-                    logger.error("Unexpected exception while returning job-" + item.getContentId() + " to queue", e2);
-                }
-
-                try {
-                    job.setExecutingMsid(null);
-                    _jobDao.update(job.getId(), job);
-                } catch (Exception e3) {
-                    logger.warn("Unexpected exception while update job-" + item.getContentId() + " msid for bookkeeping");
-                }
-            }
+        try {
+            scheduleExecution(job);
+        } catch (RejectedExecutionException e) {
+            logger.warn("Execution for job-" + job.getId() + " is rejected, return it to the queue for next turn");
+            clearExecutingMsid(job, item);
+            returnItemToQueue(item);
+        }
+    }
 
-        } else {
-            if (logger.isDebugEnabled()) {
-                logger.debug("Unable to find related job for queue item: " + item.toString());
-            }
+    private boolean returnItemToQueue(SyncQueueItemVO item) {
+        try {
+            _queueMgr.returnItem(item.getId());
+            return true;
+        } catch (Throwable thr) {
+            logger.error("Unexpected exception while returning job-" + item.getContentId() + " to queue", thr);
+            return false;
+        }
+    }
 
-            _queueMgr.purgeItem(item.getId());
+    private void clearExecutingMsid(AsyncJobVO job, SyncQueueItemVO item) {
+        try {
+            job.setExecutingMsid(null);
+            _jobDao.update(job.getId(), job);
+        } catch (Exception e) {
+            logger.warn("Unexpected exception while update job-" + item.getContentId() + " msid for bookkeeping");
         }
     }
 

framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplExecuteQueueItemTest.java

@@ -0,0 +1,129 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.cloudstack.framework.jobs.impl;
+
+import java.util.concurrent.RejectedExecutionException;
+
+import org.apache.cloudstack.framework.jobs.dao.AsyncJobDao;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InOrder;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import com.cloud.utils.exception.CloudRuntimeException;
+
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class AsyncJobManagerImplExecuteQueueItemTest {
+
+    @Mock
+    AsyncJobDao _jobDao;
+    @Mock
+    SyncQueueManager _queueMgr;
+
+    @Spy
+    @InjectMocks
+    AsyncJobManagerImpl asyncJobManager = new AsyncJobManagerImpl();
+
+    @Test
+    public void executeQueueItemDoesNotScheduleWhenTheJobUpdateFailsAndItemIsReturned() {
+        long contentId = 10L;
+        long itemId = 20L;
+        long jobId = 1L;
+
+        SyncQueueItemVO item = Mockito.mock(SyncQueueItemVO.class);
+        Mockito.when(item.getContentId()).thenReturn(contentId);
+        Mockito.when(item.getId()).thenReturn(itemId);
+
+        AsyncJobVO job = Mockito.mock(AsyncJobVO.class);
+        Mockito.when(job.getId()).thenReturn(jobId);
+        Mockito.when(_jobDao.findById(contentId)).thenReturn(job);
+
+        // Simulate the DB deadlock the catch block was written to survive.
+        Mockito.doThrow(new CloudRuntimeException("simulated DB deadlock"))
+                .when(_jobDao).update(Mockito.anyLong(), Mockito.any(AsyncJobVO.class));
+
+        // Stub the executor path so we can assert whether it is reached (and avoid the real submit).
+        Mockito.doNothing().when(asyncJobManager).scheduleExecution(Mockito.any(AsyncJobVO.class));
+
+        asyncJobManager.executeQueueItem(item, false);
+
+        // The queue item was returned for a later retry; the job must NOT also be scheduled now, or it
+        // would run twice (once here and once when the heartbeat re-dequeues the returned item).
+        Mockito.verify(_queueMgr).returnItem(itemId);
+        Mockito.verify(asyncJobManager, Mockito.never()).scheduleExecution(Mockito.any(AsyncJobVO.class));
+    }
+
+    @Test
+    public void executeQueueItemClearsExecutingMsidBeforeReturningItemOnRejectedExecution() {
+        long contentId = 10L;
+        long itemId = 20L;
+        long jobId = 1L;
+
+        SyncQueueItemVO item = Mockito.mock(SyncQueueItemVO.class);
+        Mockito.when(item.getContentId()).thenReturn(contentId);
+        Mockito.when(item.getId()).thenReturn(itemId);
+
+        AsyncJobVO job = Mockito.mock(AsyncJobVO.class);
+        Mockito.when(job.getId()).thenReturn(jobId);
+        Mockito.when(_jobDao.findById(contentId)).thenReturn(job);
+
+        // Dispatch stamps the executing msid successfully, then the executor rejects the job.
+        Mockito.doThrow(new RejectedExecutionException("pool full"))
+                .when(asyncJobManager).scheduleExecution(Mockito.any(AsyncJobVO.class));
+
+        asyncJobManager.executeQueueItem(item, false);
+
+        // The executing msid must be cleared while we still hold the item, before it goes back on the
+        // queue. Otherwise another management server could dequeue the item and stamp its own msid in
+        // the gap, and the clearing update would wipe it.
+        InOrder inOrder = Mockito.inOrder(job, _queueMgr);
+        inOrder.verify(job).setExecutingMsid(null);
+        inOrder.verify(_queueMgr).returnItem(itemId);
+    }
+
+    @Test
+    public void executeQueueItemStillSchedulesWhenRequeueAlsoFailsSoJobIsNotStranded() {
+        long contentId = 10L;
+        long itemId = 20L;
+        long jobId = 1L;
+
+        SyncQueueItemVO item = Mockito.mock(SyncQueueItemVO.class);
+        Mockito.when(item.getContentId()).thenReturn(contentId);
+        Mockito.when(item.getId()).thenReturn(itemId);
+
+        AsyncJobVO job = Mockito.mock(AsyncJobVO.class);
+        Mockito.when(job.getId()).thenReturn(jobId);
+        Mockito.when(_jobDao.findById(contentId)).thenReturn(job);
+
+        // Stamping the executing msid fails (the DB deadlock path), and returning the item also fails.
+        Mockito.doThrow(new CloudRuntimeException("simulated DB deadlock"))
+                .when(_jobDao).update(Mockito.anyLong(), Mockito.any(AsyncJobVO.class));
+        Mockito.doThrow(new CloudRuntimeException("queue unavailable"))
+                .when(_queueMgr).returnItem(Mockito.anyLong());
+        Mockito.doNothing().when(asyncJobManager).scheduleExecution(Mockito.any(AsyncJobVO.class));
+
+        asyncJobManager.executeQueueItem(item, false);
+
+        // The requeue failed, so the job must still be scheduled once instead of sitting until cleanup.
+        Mockito.verify(_queueMgr).returnItem(itemId);
+        Mockito.verify(asyncJobManager).scheduleExecution(job);
+    }
+}