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

@@ -561,7 +561,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);
     }
 
@@ -739,7 +739,7 @@ private int getAndResetPendingSignals(AsyncJob job) {
         return signals;
     }
 
-    private void executeQueueItem(SyncQueueItemVO item, boolean fromPreviousSession) {
+    protected void executeQueueItem(SyncQueueItemVO item, boolean fromPreviousSession) {
         AsyncJobVO job = _jobDao.findById(item.getContentId());
         if (job != null) {
             if (logger.isDebugEnabled()) {
@@ -764,6 +764,7 @@ private void executeQueueItem(SyncQueueItemVO item, boolean fromPreviousSession)
                 } catch (Throwable thr) {
                     logger.error("Unexpected exception while returning job-" + item.getContentId() + " to queue", thr);
                 }
+                return;
             }
 
             try {

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

@@ -0,0 +1,70 @@
+// 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 org.apache.cloudstack.framework.jobs.dao.AsyncJobDao;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+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));
+    }
+}