|
| 1 | +/* |
| 2 | + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one |
| 3 | + * or more contributor license agreements. Licensed under the Elastic License |
| 4 | + * 2.0 and the Server Side Public License, v 1; you may not use this file except |
| 5 | + * in compliance with, at your election, the Elastic License 2.0 or the Server |
| 6 | + * Side Public License, v 1. |
| 7 | + */ |
| 8 | + |
| 9 | +package org.elasticsearch.http; |
| 10 | + |
| 11 | +import org.apache.http.client.methods.HttpGet; |
| 12 | +import org.elasticsearch.action.admin.indices.recovery.RecoveryAction; |
| 13 | +import org.elasticsearch.action.admin.indices.recovery.TransportRecoveryAction; |
| 14 | +import org.elasticsearch.action.admin.indices.recovery.TransportRecoveryActionHelper; |
| 15 | +import org.elasticsearch.action.support.PlainActionFuture; |
| 16 | +import org.elasticsearch.client.Cancellable; |
| 17 | +import org.elasticsearch.client.Request; |
| 18 | +import org.elasticsearch.client.Response; |
| 19 | +import org.elasticsearch.client.ResponseListener; |
| 20 | +import org.elasticsearch.common.lease.Releasable; |
| 21 | +import org.elasticsearch.common.lease.Releasables; |
| 22 | +import org.elasticsearch.tasks.CancellableTask; |
| 23 | +import org.elasticsearch.tasks.TaskInfo; |
| 24 | +import org.elasticsearch.transport.TransportService; |
| 25 | + |
| 26 | +import java.util.ArrayList; |
| 27 | +import java.util.List; |
| 28 | +import java.util.concurrent.CancellationException; |
| 29 | +import java.util.concurrent.Semaphore; |
| 30 | + |
| 31 | +import static org.hamcrest.Matchers.empty; |
| 32 | +import static org.hamcrest.Matchers.not; |
| 33 | + |
| 34 | +public class IndicesRecoveryRestCancellationIT extends HttpSmokeTestCase { |
| 35 | + |
| 36 | + public void testIndicesRecoveryRestCancellation() throws Exception { |
| 37 | + runTest(new Request(HttpGet.METHOD_NAME, "/_recovery")); |
| 38 | + } |
| 39 | + |
| 40 | + public void testCatRecoveryRestCancellation() throws Exception { |
| 41 | + runTest(new Request(HttpGet.METHOD_NAME, "/_cat/recovery")); |
| 42 | + } |
| 43 | + |
| 44 | + private void runTest(Request request) throws Exception { |
| 45 | + |
| 46 | + createIndex("test"); |
| 47 | + ensureGreen("test"); |
| 48 | + |
| 49 | + final List<Semaphore> operationBlocks = new ArrayList<>(); |
| 50 | + for (final TransportRecoveryAction transportRecoveryAction : internalCluster().getInstances(TransportRecoveryAction.class)) { |
| 51 | + final Semaphore operationBlock = new Semaphore(1); |
| 52 | + operationBlocks.add(operationBlock); |
| 53 | + TransportRecoveryActionHelper.setOnShardOperation(transportRecoveryAction, () -> { |
| 54 | + try { |
| 55 | + operationBlock.acquire(); |
| 56 | + } catch (InterruptedException e) { |
| 57 | + throw new AssertionError(e); |
| 58 | + } |
| 59 | + operationBlock.release(); |
| 60 | + }); |
| 61 | + } |
| 62 | + assertThat(operationBlocks, not(empty())); |
| 63 | + |
| 64 | + final List<Releasable> releasables = new ArrayList<>(); |
| 65 | + try { |
| 66 | + for (final Semaphore operationBlock : operationBlocks) { |
| 67 | + operationBlock.acquire(); |
| 68 | + releasables.add(operationBlock::release); |
| 69 | + } |
| 70 | + |
| 71 | + final PlainActionFuture<Void> future = new PlainActionFuture<>(); |
| 72 | + logger.info("--> sending request"); |
| 73 | + final Cancellable cancellable = getRestClient().performRequestAsync(request, new ResponseListener() { |
| 74 | + @Override |
| 75 | + public void onSuccess(Response response) { |
| 76 | + future.onResponse(null); |
| 77 | + } |
| 78 | + |
| 79 | + @Override |
| 80 | + public void onFailure(Exception exception) { |
| 81 | + future.onFailure(exception); |
| 82 | + } |
| 83 | + }); |
| 84 | + |
| 85 | + logger.info("--> waiting for task to start"); |
| 86 | + assertBusy(() -> { |
| 87 | + final List<TaskInfo> tasks = client().admin().cluster().prepareListTasks().get().getTasks(); |
| 88 | + assertTrue(tasks.toString(), tasks.stream().anyMatch(t -> t.getAction().startsWith(RecoveryAction.NAME))); |
| 89 | + }); |
| 90 | + |
| 91 | + logger.info("--> waiting for at least one task to hit a block"); |
| 92 | + assertBusy(() -> assertTrue(operationBlocks.stream().anyMatch(Semaphore::hasQueuedThreads))); |
| 93 | + |
| 94 | + logger.info("--> cancelling request"); |
| 95 | + cancellable.cancel(); |
| 96 | + expectThrows(CancellationException.class, future::actionGet); |
| 97 | + |
| 98 | + logger.info("--> checking that all tasks are marked as cancelled"); |
| 99 | + assertBusy(() -> { |
| 100 | + boolean foundTask = false; |
| 101 | + for (TransportService transportService : internalCluster().getInstances(TransportService.class)) { |
| 102 | + for (CancellableTask cancellableTask : transportService.getTaskManager().getCancellableTasks().values()) { |
| 103 | + if (cancellableTask.getAction().startsWith(RecoveryAction.NAME)) { |
| 104 | + foundTask = true; |
| 105 | + assertTrue("task " + cancellableTask.getId() + " not cancelled", cancellableTask.isCancelled()); |
| 106 | + } |
| 107 | + } |
| 108 | + } |
| 109 | + assertTrue("found no cancellable tasks", foundTask); |
| 110 | + }); |
| 111 | + } finally { |
| 112 | + Releasables.close(releasables); |
| 113 | + } |
| 114 | + |
| 115 | + logger.info("--> checking that all tasks have finished"); |
| 116 | + assertBusy(() -> { |
| 117 | + final List<TaskInfo> tasks = client().admin().cluster().prepareListTasks().get().getTasks(); |
| 118 | + assertTrue(tasks.toString(), tasks.stream().noneMatch(t -> t.getAction().startsWith(RecoveryAction.NAME))); |
| 119 | + }); |
| 120 | + } |
| 121 | + |
| 122 | +} |
0 commit comments