Usage server should takeover immediately if the other Usage server has been stopped gracefully (#12507)

engine/schema/src/main/java/com/cloud/usage/dao/UsageJobDao.java

@@ -28,6 +28,8 @@ public interface UsageJobDao extends GenericDao<UsageJobVO, Long> {
 
     UsageJobVO getLastJob();
 
+    UsageJobVO getNextRecurringJob();
+
     UsageJobVO getNextImmediateJob();
 
     long getLastJobSuccessDateMillis();

engine/schema/src/main/java/com/cloud/usage/dao/UsageJobDaoImpl.java

@@ -156,7 +156,8 @@ public class UsageJobDaoImpl extends GenericDaoBase<UsageJobVO, Long> implements
         return jobs.get(0);
     }
 
-    private UsageJobVO getNextRecurringJob() {
+    @Override
+    public UsageJobVO getNextRecurringJob() {
         Filter filter = new Filter(UsageJobVO.class, "id", false, Long.valueOf(0), Long.valueOf(1));
         SearchCriteria<UsageJobVO> sc = createSearchCriteria();
         sc.addAnd("endMillis", SearchCriteria.Op.EQ, Long.valueOf(0));

usage/src/main/java/com/cloud/usage/UsageManagerImpl.java

@@ -2257,6 +2257,11 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
                         }
                     }
 
+                    if (_usageJobDao.getNextRecurringJob() == null) {
+                        // Own the usage processing immediately if no other node is owning it
+                        _usageJobDao.createNewJob(_hostname, _pid, UsageJobVO.JOB_TYPE_RECURRING);
+                    }
+
                     Long jobId = _usageJobDao.checkHeartbeat(_hostname, _pid, _aggregationDuration);
                     if (jobId != null) {
                         // if I'm taking over the job...see how long it's been since the last job, and if it's more than the