fix(reindex): clear stale on-demand job before triggering reindex

A leftover on-demand job from a previous run that did not complete was
preventing further reindex runs. Start the scheduler in the CLI path and
clear any existing on-demand job before triggering. The clear skips the
delete when the on-demand job is in getCurrentlyExecutingJobs(), so a
genuinely running reindex is never cleared.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mohityadav766

openmetadata-service/src/main/java/org/openmetadata/service/apps/scheduler/AppScheduler.java

@@ -336,6 +336,32 @@ public void triggerOnDemandApplication(App application, Map<String, Object> conf
     }
   }
 
+  /**
+   * Clears a persisted on-demand job/trigger for the given app from the Quartz store so a fresh
+   * on-demand trigger can be scheduled without hitting "Job is already running". Used by the CLI
+   * reindex commands to remove a leftover from a previous run that died before completing. A job
+   * that is currently executing is left untouched, so a genuinely running run is never cleared.
+   */
+  public void deleteOnDemandJob(App app) {
+    JobKey onDemandJobKey =
+        new JobKey(String.format("%s-%s", app.getName(), ON_DEMAND_JOB), APPS_JOB_GROUP);
+    try {
+      for (JobExecutionContext context : scheduler.getCurrentlyExecutingJobs()) {
+        if (context.getJobDetail().getKey().equals(onDemandJobKey)) {
+          LOG.info(
+              "On-demand job for app {} is currently executing; leaving it in place.",
+              app.getName());
+          return;
+        }
+      }
+      scheduler.deleteJob(onDemandJobKey);
+      scheduler.unscheduleJob(
+          new TriggerKey(String.format("%s-%s", app.getName(), ON_DEMAND_JOB), APPS_TRIGGER_GROUP));
+    } catch (SchedulerException ex) {
+      LOG.warn("Could not clear existing on-demand job for app {}", app.getName(), ex);
+    }
+  }
+
   public void stopApplicationRun(App application) {
     try {
       JobDetail jobDetailScheduled =

openmetadata-service/src/main/java/org/openmetadata/service/util/OpenMetadataOperations.java

@@ -1388,6 +1388,7 @@ public Integer reIndex(
       TypeRepository typeRepository = (TypeRepository) Entity.getEntityRepository(Entity.TYPE);
       TypeRegistry.instance().initialize(typeRepository);
       AppScheduler.initialize(config, collectionDAO, searchRepository);
+      AppScheduler.getInstance().start();
 
       // Prepare search repository for reindexing (e.g., initialize vector services)
       searchRepository.prepareForReindex();
@@ -1908,8 +1909,10 @@ private int executeSearchReindexApp(
       LOG.info("  - Request compression benefits (JSON payloads will be gzip compressed)");
     }
 
-    // Trigger Application
+    // Trigger Application. Clear any on-demand job left behind by a previous run that died
+    // before completing so this run is not rejected with "Job is already running".
     long currentTime = System.currentTimeMillis();
+    AppScheduler.getInstance().deleteOnDemandJob(app);
     AppScheduler.getInstance().triggerOnDemandApplication(app, JsonUtils.getMap(config));
 
     int result = waitAndReturnReindexingAppStatus(app, currentTime, progressMonitor);
@@ -1966,6 +1969,7 @@ public Integer reIndexDI(
       CollectionRegistry.getInstance().loadSeedData(jdbi, config, null, null, null, true);
       ApplicationHandler.initialize(config);
       AppScheduler.initialize(config, collectionDAO, searchRepository);
+      AppScheduler.getInstance().start();
       return executeDataInsightsReindexApp(
           batchSize, recreateIndexes, getBackfillConfiguration(startDate, endDate));
     } catch (Exception e) {
@@ -2000,8 +2004,10 @@ private int executeDataInsightsReindexApp(
             .withRecreateDataAssetsIndex(recreateIndexes)
             .withBackfillConfiguration(backfillConfiguration);
 
-    // Trigger Application
+    // Trigger Application. Clear any on-demand job left behind by a previous run that died
+    // before completing so this run is not rejected with "Job is already running".
     long currentTime = System.currentTimeMillis();
+    AppScheduler.getInstance().deleteOnDemandJob(app);
     AppScheduler.getInstance().triggerOnDemandApplication(app, JsonUtils.getMap(config));
     return waitAndReturnReindexingAppStatus(app, currentTime);
   }

openmetadata-service/src/test/java/org/openmetadata/service/apps/scheduler/AppSchedulerTest.java

@@ -1,5 +1,6 @@
 package org.openmetadata.service.apps.scheduler;
 
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -19,6 +20,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -34,8 +36,11 @@
 import org.quartz.JobDetail;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobKey;
+import org.quartz.ObjectAlreadyExistsException;
 import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
 import org.quartz.Trigger;
+import org.quartz.TriggerKey;
 import sun.misc.Unsafe;
 
 @ExtendWith(MockitoExtension.class)
@@ -287,4 +292,137 @@ void testConcurrentApp_twoParallelJobs_differentIdentities() throws Exception {
     assertTrue(identity1.contains(workflowId1));
     assertTrue(identity2.contains(workflowId2));
   }
+
+  // --- Tests for the reindex/ops path: clear a stale on-demand job, then trigger ---
+
+  private App nonConcurrentApp(String name) {
+    return new App()
+        .withId(UUID.randomUUID())
+        .withName(name)
+        .withFullyQualifiedName(name)
+        .withClassName("org.openmetadata.service.resources.apps.TestApp")
+        .withAllowConcurrentExecution(false)
+        .withRuntime(new ScheduledExecutionContext().withEnabled(true));
+  }
+
+  @Test
+  void testDeleteOnDemandJob_removesOnDemandJobAndTrigger() throws Exception {
+    AppScheduler appScheduler = createSchedulerWithMock();
+    App app = nonConcurrentApp("SearchIndexingApplication");
+
+    appScheduler.deleteOnDemandJob(app);
+
+    String onDemandIdentity =
+        String.format("SearchIndexingApplication-%s", AppScheduler.ON_DEMAND_JOB);
+    verify(mockScheduler).deleteJob(new JobKey(onDemandIdentity, AppScheduler.APPS_JOB_GROUP));
+    verify(mockScheduler)
+        .unscheduleJob(new TriggerKey(onDemandIdentity, AppScheduler.APPS_TRIGGER_GROUP));
+  }
+
+  @Test
+  void testDeleteOnDemandJob_skipsWhenJobCurrentlyExecuting() throws Exception {
+    AppScheduler appScheduler = createSchedulerWithMock();
+    App app = nonConcurrentApp("SearchIndexingApplication");
+    String onDemandIdentity =
+        String.format("SearchIndexingApplication-%s", AppScheduler.ON_DEMAND_JOB);
+    JobKey onDemandKey = new JobKey(onDemandIdentity, AppScheduler.APPS_JOB_GROUP);
+
+    // The on-demand job is genuinely executing right now.
+    JobDetail running =
+        JobBuilder.newJob(Job.class)
+            .withIdentity(onDemandIdentity, AppScheduler.APPS_JOB_GROUP)
+            .build();
+    JobExecutionContext execContext = mock(JobExecutionContext.class);
+    when(execContext.getJobDetail()).thenReturn(running);
+    when(mockScheduler.getCurrentlyExecutingJobs()).thenReturn(List.of(execContext));
+
+    appScheduler.deleteOnDemandJob(app);
+
+    // A running job must never be cleared.
+    verify(mockScheduler, never()).deleteJob(onDemandKey);
+    verify(mockScheduler, never()).unscheduleJob(any(TriggerKey.class));
+  }
+
+  @Test
+  void testDeleteOnDemandJob_isBestEffortOnSchedulerException() throws Exception {
+    AppScheduler appScheduler = createSchedulerWithMock();
+    App app = nonConcurrentApp("SearchIndexingApplication");
+    when(mockScheduler.deleteJob(any(JobKey.class)))
+        .thenThrow(new SchedulerException("store unavailable"));
+
+    // Clearing a leftover is best-effort; a failure here must not break the reindex flow.
+    assertDoesNotThrow(() -> appScheduler.deleteOnDemandJob(app));
+  }
+
+  @Test
+  void testStaleOnDemandJob_blocksTriggerWithoutClear() throws Exception {
+    AppScheduler appScheduler = createSchedulerWithMock();
+    App app = nonConcurrentApp("SearchIndexingApplication");
+    String onDemandIdentity =
+        String.format("SearchIndexingApplication-%s", AppScheduler.ON_DEMAND_JOB);
+    JobKey onDemandKey = new JobKey(onDemandIdentity, AppScheduler.APPS_JOB_GROUP);
+
+    // A leftover on-demand job persisted by a previous run that died (not currently running).
+    JobDetail stale =
+        JobBuilder.newJob(Job.class)
+            .withIdentity(onDemandIdentity, AppScheduler.APPS_JOB_GROUP)
+            .build();
+    when(mockScheduler.getJobDetail(
+            new JobKey("SearchIndexingApplication", AppScheduler.APPS_JOB_GROUP)))
+        .thenReturn(null);
+    when(mockScheduler.getJobDetail(onDemandKey)).thenReturn(stale);
+    when(mockScheduler.getCurrentlyExecutingJobs()).thenReturn(Collections.emptyList());
+    // Quartz rejects scheduling a duplicate key while the stale job is still persisted.
+    when(mockScheduler.scheduleJob(any(JobDetail.class), any(Trigger.class)))
+        .thenThrow(new ObjectAlreadyExistsException("duplicate job key"));
+
+    assertThrows(
+        UnhandledServerException.class,
+        () -> appScheduler.triggerOnDemandApplication(app, new HashMap<>()));
+  }
+
+  @Test
+  void testReindexPath_clearStaleThenTriggerSucceeds() throws Exception {
+    AppScheduler appScheduler = createSchedulerWithMock();
+    App app = nonConcurrentApp("SearchIndexingApplication");
+    String onDemandIdentity =
+        String.format("SearchIndexingApplication-%s", AppScheduler.ON_DEMAND_JOB);
+    JobKey onDemandKey = new JobKey(onDemandIdentity, AppScheduler.APPS_JOB_GROUP);
+
+    // Stale leftover present until deleteOnDemandJob removes it, mirroring the JDBC job store.
+    AtomicBoolean stalePresent = new AtomicBoolean(true);
+    JobDetail stale =
+        JobBuilder.newJob(Job.class)
+            .withIdentity(onDemandIdentity, AppScheduler.APPS_JOB_GROUP)
+            .build();
+    when(mockScheduler.getJobDetail(
+            new JobKey("SearchIndexingApplication", AppScheduler.APPS_JOB_GROUP)))
+        .thenReturn(null);
+    when(mockScheduler.getJobDetail(onDemandKey))
+        .thenAnswer(inv -> stalePresent.get() ? stale : null);
+    when(mockScheduler.getCurrentlyExecutingJobs()).thenReturn(Collections.emptyList());
+    when(mockScheduler.deleteJob(onDemandKey))
+        .thenAnswer(
+            inv -> {
+              stalePresent.set(false);
+              return true;
+            });
+    when(mockScheduler.scheduleJob(any(JobDetail.class), any(Trigger.class)))
+        .thenAnswer(
+            inv -> {
+              if (stalePresent.get()) {
+                throw new ObjectAlreadyExistsException("duplicate job key");
+              }
+              return null;
+            });
+
+    // The reindex CLI path: clear the leftover, then trigger.
+    appScheduler.deleteOnDemandJob(app);
+    appScheduler.triggerOnDemandApplication(app, new HashMap<>());
+
+    verify(mockScheduler).deleteJob(onDemandKey);
+    ArgumentCaptor<JobDetail> jobCaptor = ArgumentCaptor.forClass(JobDetail.class);
+    verify(mockScheduler).scheduleJob(jobCaptor.capture(), any(Trigger.class));
+    assertEquals(onDemandIdentity, jobCaptor.getValue().getKey().getName());
+  }
 }