[ML] Fix timeout bug in DBQ deletion of unused and orphan ML data (elastic#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.

Author: valeriy42

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: []

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobDataDeleter.java

@@ -109,15 +109,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;
         }
 
@@ -134,7 +126,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)
@@ -147,6 +144,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
@@ -311,7 +318,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)
@@ -504,7 +511,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()))