From 19791f02511280459d0476604bb7d74112cb0bf7 Mon Sep 17 00:00:00 2001
From: Valeriy Khakhutskyy <1292899+valeriy42@users.noreply.github.com>
Date: Thu, 26 Jun 2025 13:24:59 +0200
Subject: [PATCH] [ML] Fix timeout bug in DBQ deletion of unused and orphan ML
 data (#130083)

There was a bug in the code for deleting unused and orphan ML data. When deletion using DBQ occurred, the bug caused the request to time out. This PR resolves the issue.
---
 docs/changelog/130083.yaml                    |  5 +++
 .../ml/job/persistence/JobDataDeleter.java    | 36 ++++++++++++-------
 2 files changed, 29 insertions(+), 12 deletions(-)
 create mode 100644 docs/changelog/130083.yaml

diff --git a/docs/changelog/130083.yaml b/docs/changelog/130083.yaml
new file mode 100644
index 0000000000000..1b32881659531
--- /dev/null
+++ b/docs/changelog/130083.yaml
@@ -0,0 +1,5 @@
+pr: 130083
+summary: Fix timeout bug in DBQ deletion of unused and orphan ML data
+area: Machine Learning
+type: bug
+issues: []
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobDataDeleter.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobDataDeleter.java
index af57729cdae94..5929d0ff74c8f 100644
--- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobDataDeleter.java
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobDataDeleter.java
@@ -107,15 +107,7 @@ public JobDataDeleter(Client client, String jobId, boolean deleteUserAnnotations
      */
     public void deleteModelSnapshots(List<ModelSnapshot> modelSnapshots, ActionListener<BulkByScrollResponse> listener) {
         if (modelSnapshots.isEmpty()) {
-            listener.onResponse(
-                new BulkByScrollResponse(
-                    TimeValue.ZERO,
-                    new BulkByScrollTask.Status(Collections.emptyList(), null),
-                    Collections.emptyList(),
-                    Collections.emptyList(),
-                    false
-                )
-            );
+            listener.onResponse(emptyBulkByScrollResponse());
             return;
         }
 
@@ -132,7 +124,12 @@ public void deleteModelSnapshots(List<ModelSnapshot> modelSnapshots, ActionListe
             indices.add(AnomalyDetectorsIndex.jobResultsAliasedName(modelSnapshot.getJobId()));
         }
 
-        String[] indicesToQuery = removeReadOnlyIndices(new ArrayList<>(indices), listener, "model snapshots", null);
+        String[] indicesToQuery = removeReadOnlyIndices(
+            new ArrayList<>(indices),
+            listener,
+            "model snapshots",
+            () -> listener.onResponse(emptyBulkByScrollResponse())
+        );
         if (indicesToQuery.length == 0) return;
 
         DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(indicesToQuery).setRefresh(true)
@@ -145,6 +142,16 @@ public void deleteModelSnapshots(List<ModelSnapshot> modelSnapshots, ActionListe
         executeAsyncWithOrigin(client, ML_ORIGIN, DeleteByQueryAction.INSTANCE, deleteByQueryRequest, listener);
     }
 
+    private static BulkByScrollResponse emptyBulkByScrollResponse() {
+        return new BulkByScrollResponse(
+            TimeValue.ZERO,
+            new BulkByScrollTask.Status(Collections.emptyList(), null),
+            Collections.emptyList(),
+            Collections.emptyList(),
+            false
+        );
+    }
+
     /**
      * Asynchronously delete the annotations
      * If the deleteUserAnnotations field is set to true then all
@@ -309,7 +316,7 @@ public void deleteDatafeedTimingStats(ActionListener<BulkByScrollResponse> liste
             List.of(AnomalyDetectorsIndex.jobResultsAliasedName(jobId)),
             listener,
             "datafeed timing stats",
-            null
+            () -> listener.onResponse(emptyBulkByScrollResponse())
         );
         if (indicesToQuery.length == 0) return;
         DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(indicesToQuery).setRefresh(true)
@@ -502,7 +509,12 @@ private void deleteResultsByQuery(
         ActionListener<BroadcastResponse> refreshListener = ActionListener.wrap(refreshResponse -> {
             logger.info("[{}] running delete by query on [{}]", jobId, String.join(", ", indices));
             ConstantScoreQueryBuilder query = new ConstantScoreQueryBuilder(new TermQueryBuilder(Job.ID.getPreferredName(), jobId));
-            String[] indicesToQuery = removeReadOnlyIndices(List.of(indices), listener, "results", null);
+            String[] indicesToQuery = removeReadOnlyIndices(
+                List.of(indices),
+                listener,
+                "results",
+                () -> listener.onResponse(emptyBulkByScrollResponse())
+            );
             if (indicesToQuery.length == 0) return;
             DeleteByQueryRequest request = new DeleteByQueryRequest(indicesToQuery).setQuery(query)
                 .setIndicesOptions(MlIndicesUtils.addIgnoreUnavailable(IndicesOptions.lenientExpandOpenHidden()))