Skip to content

Fix 1.x compatibility bug with stored Tasks #5412

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
andrross merged 1 commit into from
Nov 30, 2022
Merged

Fix 1.x compatibility bug with stored Tasks #5412

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
### Deprecated
### Removed
### Fixed
- Fix 1.x compatibility bug with stored Tasks ([#5412](https://github.com/opensearch-project/OpenSearch/pull/5412))
### Security

[Unreleased 3.0]: https://github.com/opensearch-project/OpenSearch/compare/2.4...HEAD
Expand Down
53 changes: 32 additions & 21 deletions qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/SystemIndicesUpgradeIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,14 +32,16 @@

package org.opensearch.upgrades;

import org.opensearch.LegacyESVersion;
import org.opensearch.Version;
import org.hamcrest.MatcherAssert;
import org.opensearch.client.Request;
import org.opensearch.client.Response;
import org.opensearch.client.ResponseException;
import org.opensearch.test.XContentTestUtils.JsonMapView;

import java.io.IOException;
import java.util.Map;

import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasKey;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
Expand All @@ -62,25 +64,7 @@ public void testSystemIndicesUpgrades() throws Exception {
"{\"f1\": \"v1\", \"f2\": \"v2\"}\n");
client().performRequest(bulk);

// start a async reindex job
Request reindex = new Request("POST", "/_reindex");
reindex.setJsonEntity(
"{\n" +
" \"source\":{\n" +
" \"index\":\"test_index_old\"\n" +
" },\n" +
" \"dest\":{\n" +
" \"index\":\"test_index_reindex\"\n" +
" }\n" +
"}");
reindex.addParameter("wait_for_completion", "false");
Map<String, Object> response = entityAsMap(client().performRequest(reindex));
String taskId = (String) response.get("task");

// wait for task
Request getTask = new Request("GET", "/_tasks/" + taskId);
getTask.addParameter("wait_for_completion", "true");
client().performRequest(getTask);
createAndVerifyStoredTask();

// make sure .tasks index exists
Request getTasksIndex = new Request("GET", "/.tasks");
Expand All @@ -97,6 +81,8 @@ public void testSystemIndicesUpgrades() throws Exception {
}
});
} else if (CLUSTER_TYPE == ClusterType.UPGRADED) {
createAndVerifyStoredTask();

assertBusy(() -> {
Request clusterStateRequest = new Request("GET", "/_cluster/state/metadata");
Map<String, Object> indices = new JsonMapView(entityAsMap(client().performRequest(clusterStateRequest)))
Expand All @@ -115,4 +101,29 @@ public void testSystemIndicesUpgrades() throws Exception {
});
}
}

/**
* Completed tasks get persisted into the .tasks index, so this method waits
* until the task is completed in order to verify that it has been successfully
* written to the index and can be retrieved.
*/
private static void createAndVerifyStoredTask() throws Exception {
// Use update by query to create an async task
final Request updateByQueryRequest = new Request("POST", "/test_index_old/_update_by_query");

@reta reta Nov 30, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just curious, _reindex was replaced with _update_by_query, is there some particular reason?

@andrross andrross Nov 30, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because it is done twice, once for the old cluster and again for the updated cluster. Reindex creates a new index so you can't make the exact same request twice (unless you closed or deleted the new index). Update by query is simpler because it takes no parameters and doesn't create any new resources :)

@andrross andrross Nov 30, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Previously this step was only being done on the old cluster, so there was no problem. My change does this test for both the old and updated clusters.)

updateByQueryRequest.addParameter("wait_for_completion", "false");
final Response updateByQueryResponse = client().performRequest(updateByQueryRequest);
MatcherAssert.assertThat(updateByQueryResponse.getStatusLine().getStatusCode(), equalTo(200));
final String taskId = (String) entityAsMap(updateByQueryResponse).get("task");

// wait for task to complete
waitUntil(() -> {
try {
final Response getTaskResponse = client().performRequest(new Request("GET", "/_tasks/" + taskId));
MatcherAssert.assertThat(getTaskResponse.getStatusLine().getStatusCode(), equalTo(200));
return (Boolean) entityAsMap(getTaskResponse).get("completed");
} catch (IOException e) {
throw new AssertionError(e);
}
});
}
}
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/tasks/TaskResultsService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ public class TaskResultsService {

public static final String TASK_RESULT_MAPPING_VERSION_META_FIELD = "version";

public static final int TASK_RESULT_MAPPING_VERSION = 3; // must match version in task-index-mapping.json
public static final int TASK_RESULT_MAPPING_VERSION = 4; // must match version in task-index-mapping.json

/**
* The backoff policy to use when saving a task result fails. The total wait
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/resources/org/opensearch/tasks/task-index-mapping.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
{
"_doc" : {
"_meta": {
"version": 3
"version": 4
},
"dynamic" : "strict",
"properties" : {
Expand Down