Usage: Heartbeat should not schedule usage job when a job is already running (apache#12616)

abh1sar · Locharla, Sandeep · commit f56fdcbdceb5 · 2026-04-09T22:16:12.000+05:30
diff --git a/usage/src/main/java/com/cloud/usage/UsageManagerImpl.java b/usage/src/main/java/com/cloud/usage/UsageManagerImpl.java
@@ -31,6 +31,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.cloud.network.Network;
 import com.cloud.usage.dao.UsageNetworksDao;
@@ -179,6 +180,7 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
     private final List<UsageVmDiskVO> usageVmDisks = new ArrayList<UsageVmDiskVO>();
 
     private final ScheduledExecutorService _executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-Job"));
+    private final AtomicBoolean isParsingJobRunning = new AtomicBoolean(false);
     private final ScheduledExecutorService _heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-HB"));
     private final ScheduledExecutorService _sanityExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Usage-Sanity"));
     private Future _scheduledFuture = null;
@@ -354,7 +356,12 @@ public void run() {
         (new ManagedContextRunnable() {
             @Override
             protected void runInContext() {
-                runInContextInternal();
+                isParsingJobRunning.set(true);
+                try {
+                    runInContextInternal();
+                } finally {
+                    isParsingJobRunning.set(false);
+                }
             }
         }).run();
     }
@@ -2177,9 +2184,14 @@ protected void runInContext() {
 
                         if ((timeSinceLastSuccessJob > 0) && (timeSinceLastSuccessJob > (aggregationDurationMillis - 100))) {
                             if (timeToJob > (aggregationDurationMillis / 2)) {
-                                logger.debug("it's been {} ms since last usage job and {} ms until next job, scheduling an immediate job to catch up (aggregation duration is {} minutes)"
-                                    , timeSinceLastSuccessJob, timeToJob, _aggregationDuration);
-                                scheduleParse();
+                                logger.debug("Heartbeat: it's been {} ms since last finished usage job and {} ms until next job (aggregation duration is {} minutes)",
+                                        timeSinceLastSuccessJob, timeToJob, _aggregationDuration);
+                                if (isParsingJobRunning.get()) {
+                                    logger.debug("Heartbeat: A parsing job is already running");
+                                } else {
+                                    logger.debug("Heartbeat: Scheduling an immediate job to catch up");
+                                    scheduleParse();
+                                }
                             }
                         }
 
