|
6 | 6 | package org.elasticsearch.xpack.enrich;
|
7 | 7 |
|
8 | 8 | import org.apache.lucene.search.TotalHits;
|
| 9 | +import org.elasticsearch.ResourceNotFoundException; |
| 10 | +import org.elasticsearch.action.ActionFuture; |
| 11 | +import org.elasticsearch.action.admin.cluster.node.tasks.get.GetTaskResponse; |
9 | 12 | import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
|
10 | 13 | import org.elasticsearch.action.bulk.BulkItemResponse;
|
11 | 14 | import org.elasticsearch.action.bulk.BulkRequest;
|
|
23 | 26 | import org.elasticsearch.index.reindex.ReindexPlugin;
|
24 | 27 | import org.elasticsearch.ingest.common.IngestCommonPlugin;
|
25 | 28 | import org.elasticsearch.plugins.Plugin;
|
| 29 | +import org.elasticsearch.tasks.TaskInfo; |
26 | 30 | import org.elasticsearch.test.ESIntegTestCase;
|
27 | 31 | import org.elasticsearch.xpack.core.XPackSettings;
|
28 | 32 | import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
|
@@ -222,7 +226,27 @@ private static void createAndExecutePolicy() {
|
222 | 226 | );
|
223 | 227 | PutEnrichPolicyAction.Request request = new PutEnrichPolicyAction.Request(POLICY_NAME, enrichPolicy);
|
224 | 228 | client().execute(PutEnrichPolicyAction.INSTANCE, request).actionGet();
|
225 |
| - client().execute(ExecuteEnrichPolicyAction.INSTANCE, new ExecuteEnrichPolicyAction.Request(POLICY_NAME)).actionGet(); |
| 229 | + final ActionFuture<ExecuteEnrichPolicyAction.Response> policyExecuteFuture = client().execute( |
| 230 | + ExecuteEnrichPolicyAction.INSTANCE, |
| 231 | + new ExecuteEnrichPolicyAction.Request(POLICY_NAME) |
| 232 | + ); |
| 233 | + // Make sure we can deserialize enrich policy execution task status |
| 234 | + final List<TaskInfo> tasks = client().admin() |
| 235 | + .cluster() |
| 236 | + .prepareListTasks() |
| 237 | + .setActions(EnrichPolicyExecutor.TASK_ACTION) |
| 238 | + .get() |
| 239 | + .getTasks(); |
| 240 | + // Best effort, sometimes the enrich policy task will not be visible yet or will have already finished |
| 241 | + if (tasks.isEmpty() == false) { |
| 242 | + try { |
| 243 | + final GetTaskResponse getTaskResponse = client().admin().cluster().prepareGetTask(tasks.get(0).getTaskId()).get(); |
| 244 | + assertEquals(getTaskResponse.getTask().getTask().getAction(), EnrichPolicyExecutor.TASK_ACTION); |
| 245 | + } catch (ResourceNotFoundException e) { |
| 246 | + // ignored, could be the task has already finished |
| 247 | + } |
| 248 | + } |
| 249 | + policyExecuteFuture.actionGet(); |
226 | 250 | }
|
227 | 251 |
|
228 | 252 | private static void createPipeline() {
|
|
0 commit comments