github-mirrors/elasticsearch
1
0
Fork 0
mirror of https://github.com/elastic/elasticsearch.git synced 2025-06-27 17:10:22 -04:00
Code Issues Projects Releases Packages Wiki Activity

[ML] Fix timeout bug in DBQ deletion of unused and orphan ML data (#130083)

Browse source 
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.
This commit is contained in:
Valeriy Khakhutskyy 2025-06-26 13:24:59 +02:00 • committed by GitHub
parent a6004a6067
commit 4c7d922eeb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 29 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

5
docs/changelog/130083.yaml Normal file
View file

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

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

@ -109,15 +109,7 @@ public class JobDataDeleter {
*/
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 class JobDataDeleter {
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 class JobDataDeleter {
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 class JobDataDeleter {
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 @@ public class JobDataDeleter {
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()))