diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategy.java b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategy.java index 00e1240fd395..c283ae288eda 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategy.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategy.java @@ -103,6 +103,10 @@ private ExecutionResult doExecute( int partitionSize = jobData.getPartitionSize() != null ? jobData.getPartitionSize() : 10000; distributedExecutor = new DistributedSearchIndexExecutor(collectionDAO, partitionSize); + // A stop that landed before this executor existed had nothing to stop yet; hand it on. + if (stopped.get()) { + distributedExecutor.stop(); + } distributedExecutor.performStartupRecovery(); distributedExecutor.addListener(listeners); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestrator.java b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestrator.java index 8542b9efe36e..3c03148777d1 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestrator.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestrator.java @@ -230,6 +230,10 @@ private void runReindexing() { DistributedIndexingStrategy strategy = createDistributedStrategy(); activeStrategy = strategy; + // A stop that landed before this strategy existed had nothing to stop yet; hand it on. + if (stopped) { + strategy.stop(); + } registerProgressListeners(strategy); ReindexingConfiguration config = buildReindexingConfiguration(); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutor.java b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutor.java index afb49dbb94b6..e895df6673c6 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutor.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutor.java @@ -119,7 +119,8 @@ public static boolean isCoordinatingJob(UUID jobId) { private final String serverId; private final CompositeProgressListener listeners = new CompositeProgressListener(); - @Getter private SearchIndexJob currentJob; + // Written by the reindex thread, read by stop() on the request thread that asked for the stop. + @Getter private volatile SearchIndexJob currentJob; private DistributedJobStatsAggregator statsAggregator; private ExecutorService workerExecutor; private final Set activePartitions = ConcurrentHashMap.newKeySet(); @@ -344,6 +345,9 @@ public ExecutionResult execute( } UUID jobId = currentJob.getId(); + if (stopped.get()) { + return endJobStoppedBeforeStart(jobId); + } LOG.info("Server {} starting execution of job {}", serverId, jobId); boolean startedJob = false; @@ -626,9 +630,26 @@ public ExecutionResult execute( LOG.debug("Removed job {} from coordinated jobs set", jobId); } - // Get final job state - currentJob = coordinator.getJobWithAggregatedStats(jobId); + return finalResult(jobId); + } + /** + * {@link #stop()} can land while {@link #createJob} is still building partitions, when there is + * no job yet to request a stop for. Starting the job anyway would flip it to RUNNING with none of + * our workers on it: peer servers then adopt it as an orphan and reindex what was just stopped, + * and nothing ever finishes the app run. Stop it before it starts instead. + */ + private ExecutionResult endJobStoppedBeforeStart(UUID jobId) { + LOG.info("Stop was requested before job {} started; not starting it", jobId); + if (!isJobTerminalOrStopping(jobId)) { + coordinator.requestStop(jobId); + } + coordinator.releaseReindexLock(jobId); + return finalResult(jobId); + } + + private ExecutionResult finalResult(UUID jobId) { + currentJob = coordinator.getJobWithAggregatedStats(jobId); return new ExecutionResult( currentJob.getStatus(), currentJob.getTotalRecords(), diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategyTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategyTest.java index 7ee4a8358cae..69453210b0fa 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategyTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/DistributedIndexingStrategyTest.java @@ -803,6 +803,39 @@ void executeReturnsFailedResultWhenDistributedExecutorStartupFails() { } } + /** + * Regression: a stop that lands before the executor exists finds nothing to stop. The executor + * created afterwards never heard of it, so it created and started the job the user had stopped. + */ + @Test + @SuppressWarnings({"rawtypes", "unchecked"}) + void stopBeforeExecutorExistsIsHandedToTheExecutorCreatedAfterIt() { + EntityRepository entityRepository = mock(EntityRepository.class); + EntityDAO entityDao = mock(EntityDAO.class); + + when(entityRepository.getDao()).thenReturn(entityDao); + when(entityRepository.getReindexFilter()).thenReturn(new ListFilter(Include.ALL)); + when(entityDao.listCount(any(ListFilter.class))).thenReturn(5); + + try (MockedStatic entityMock = mockStatic(Entity.class); + MockedConstruction executorConstruction = + mockConstruction( + DistributedSearchIndexExecutor.class, + (mock, context) -> + doThrow(new RuntimeException("end the run before job creation")) + .when(mock) + .performStartupRecovery())) { + entityMock.when(() -> Entity.getEntityRepository(Entity.TABLE)).thenReturn(entityRepository); + + strategy.stop(); + strategy.execute( + ReindexingConfiguration.builder().entities(Set.of(Entity.TABLE)).build(), + context(APP_ID)); + + verify(executorConstruction.constructed().getFirst()).stop(); + } + } + @Test @SuppressWarnings({"rawtypes", "unchecked"}) void executeClosesSinkAndReturnsFailedWhenDoExecuteThrowsAndSinkCloseAlsoFails() diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestratorTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestratorTest.java index bfcd12cd88e2..2ad86696b4b7 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestratorTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestratorTest.java @@ -10,6 +10,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockConstruction; @@ -308,6 +309,56 @@ void stopStopsActiveStrategyAndPushesStoppedStatus() throws Exception { assertNotNull(appRunRecord.getEndTime()); } + /** + * Regression: a stop that lands during preflight finds no strategy to stop. The strategy created + * afterwards never heard of it and ran the whole reindex the user had stopped. + */ + @Test + void stopDuringPreflightIsHandedToTheStrategyCreatedAfterIt() { + EventPublisherJob jobData = new EventPublisherJob().withEntities(Set.of(Entity.TABLE)); + EntityRepository entityRepository = mock(EntityRepository.class); + EntityDAO entityDao = mock(EntityDAO.class); + + when(context.getJobName()).thenReturn("scheduled"); + when(context.createProgressListener(jobData)) + .thenReturn(mock(ReindexingProgressListener.class)); + when(entityRepository.getDao()).thenReturn(entityDao); + when(entityDao.listCount(any())).thenReturn(5); + doAnswer( + invocation -> { + orchestrator.stop(); + return null; + }) + .when(searchRepository) + .ensureHybridSearchPipeline(); + + try (MockedStatic entityMock = mockStatic(Entity.class); + MockedStatic metricsMock = mockStatic(ReindexingMetrics.class); + MockedStatic websocketMock = mockStatic(WebSocketManager.class); + MockedConstruction ignoredCleaner = mockOrphanCleaner(); + MockedConstruction strategyConstruction = + mockConstruction( + DistributedIndexingStrategy.class, + (strategy, context1) -> + when(strategy.execute(any(), any())) + .thenReturn( + ExecutionResult.builder() + .status(ExecutionResult.Status.STOPPED) + .startTime(10L) + .endTime(20L) + .build()))) { + metricsMock.when(ReindexingMetrics::getInstance).thenReturn(null); + websocketMock.when(WebSocketManager::getInstance).thenReturn(null); + entityMock.when(() -> Entity.getEntityRepository(Entity.TABLE)).thenReturn(entityRepository); + + orchestrator.run(jobData); + + verify(strategyConstruction.constructed().getFirst()).stop(); + assertEquals(EventPublisherJob.Status.STOPPED, orchestrator.getJobData().getStatus()); + assertEquals(AppRunRecord.Status.STOPPED, appRunRecord.getStatus()); + } + } + @Test void setupEntitiesExpandsAllAndCountTotalEntitiesSkipsUnsupportedTypes() throws Exception { EventPublisherJob jobData = new EventPublisherJob().withEntities(Set.of("all", Entity.TABLE)); diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutorTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutorTest.java index da938b6ff43e..1d40a03db75f 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutorTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutorTest.java @@ -320,6 +320,38 @@ void stopStopsWorkersAndRequestsCoordinatorStopOnlyOnce() throws Exception { verify(coordinator, times(1)).requestStop(job.getId()); } + /** + * Regression: a stop that lands while {@code createJob()} is still building partitions finds no + * job to stop. {@code execute()} then started the job anyway, leaving it RUNNING with none of our + * workers on it, so peer servers adopted it as an orphan, reindexed what the user had stopped, + * and the app run stayed "running" for good. + */ + @Test + void executeEndsJobStoppedBeforeItStartedInsteadOfStartingIt() throws Exception { + UUID jobId = UUID.randomUUID(); + SearchIndexJob readyJob = + SearchIndexJob.builder().id(jobId).status(IndexJobStatus.READY).build(); + ReindexingProgressListener listener = mock(ReindexingProgressListener.class); + executor.addListener(listener); + executor.stop(); + setField("currentJob", readyJob); + when(coordinator.getJob(jobId)).thenReturn(Optional.of(readyJob)); + when(coordinator.getJobWithAggregatedStats(jobId)) + .thenReturn(readyJob.withStatus(IndexJobStatus.STOPPED)); + + DistributedSearchIndexExecutor.ExecutionResult result = + executor.execute( + mock(BulkSink.class), + stagedContext("table"), + ReindexingConfiguration.builder().entities(Set.of("table")).build()); + + assertEquals(IndexJobStatus.STOPPED, result.status()); + verify(coordinator, never()).startJob(jobId); + verify(coordinator).requestStop(jobId); + verify(coordinator).releaseReindexLock(jobId); + verify(listener, never()).onJobStarted(any()); + } + /** * Regression: clicking Stop in the UI used to "do nothing" because workers blocked inside the * bulk-sink semaphore, slow DB queries, or {@code waitForSinkOperations} (5-min deadline)