Skip to content

Commit

Permalink
Rename methods in PersistentTasksService (#30837)
Browse files Browse the repository at this point in the history 
This commit renames methods in the PersistentTasksService, to
make obvious that the methods send requests in order to change
the state of persistent tasks.

Relates to #29608.
  • Loading branch information
@tlrx
tlrx committed May 30, 2018
1 parent e2749bd commit 02b6ea3
Show file tree
Hide file tree
Showing 17 changed files with 173 additions and 165 deletions.
16 changes: 10 additions & 6 deletions server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -103,12 +103,16 @@ public long getAllocationId() {
}

/**
* Waits for this persistent task to have the desired state.
* Waits for a given persistent task to comply with a given predicate, then call back the listener accordingly.
*
* @param predicate the persistent task predicate to evaluate
* @param timeout a timeout for waiting
* @param listener the callback listener
*/
public void waitForPersistentTaskStatus(Predicate<PersistentTasksCustomMetaData.PersistentTask<?>> predicate,
@Nullable TimeValue timeout,
PersistentTasksService.WaitForPersistentTaskStatusListener<?> listener) {
persistentTasksService.waitForPersistentTaskStatus(persistentTaskId, predicate, timeout, listener);
public void waitForPersistentTask(final Predicate<PersistentTasksCustomMetaData.PersistentTask<?>> predicate,
final @Nullable TimeValue timeout,
final PersistentTasksService.WaitForPersistentTaskListener<?> listener) {
persistentTasksService.waitForPersistentTaskCondition(persistentTaskId, predicate, timeout, listener);
}

final boolean isCompleted() {
Expand Down Expand Up @@ -143,7 +147,7 @@ private void completeAndNotifyIfNeeded(@Nullable Exception failure) {
this.failure = failure;
if (prevState == State.STARTED) {
logger.trace("sending notification for completed task [{}] with id [{}]", getAction(), getPersistentTaskId());
persistentTasksService.sendCompletionNotification(getPersistentTaskId(), getAllocationId(), failure, new
persistentTasksService.sendCompletionRequest(getPersistentTaskId(), getAllocationId(), failure, new
ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
Expand Down
3 changes: 2 additions & 1 deletion server/src/main/java/org/elasticsearch/persistent/PersistentTasksNodeService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,8 @@ private void cancelTask(Long allocationId) {
AllocatedPersistentTask task = runningTasks.remove(allocationId);
if (task.markAsCancelled()) {
// Cancel the local task using the task manager
persistentTasksService.sendTaskManagerCancellation(task.getId(), new ActionListener<CancelTasksResponse>() {
String reason = "task has been removed, cancelling locally";
persistentTasksService.sendCancelRequest(task.getId(), reason, new ActionListener<CancelTasksResponse>() {
@Override
public void onResponse(CancelTasksResponse cancelTasksResponse) {
logger.trace("Persistent task [{}] with id [{}] and allocation id [{}] was cancelled", task.getAction(),
Expand Down
212 changes: 108 additions & 104 deletions server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java
View file

Large diffs are not rendered by default.

3 changes: 1 addition & 2 deletions server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorFullRestartIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,8 +65,7 @@ public void testFullClusterRestart() throws Exception {
PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
futures.add(future);
taskIds[i] = UUIDs.base64UUID();
service.startPersistentTask(taskIds[i], TestPersistentTasksExecutor.NAME, randomBoolean() ? null : new TestParams("Blah"),
future);
service.sendStartRequest(taskIds[i], TestPersistentTasksExecutor.NAME, randomBoolean() ? null : new TestParams("Blah"), future);
}

for (int i = 0; i < numberOfTasks; i++) {
Expand Down
34 changes: 17 additions & 17 deletions server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
import org.elasticsearch.tasks.TaskInfo;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
import org.elasticsearch.persistent.PersistentTasksService.WaitForPersistentTaskStatusListener;
import org.elasticsearch.persistent.PersistentTasksService.WaitForPersistentTaskListener;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.Status;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestParams;
Expand Down Expand Up @@ -69,15 +69,15 @@ public void cleanup() throws Exception {
assertNoRunningTasks();
}

public static class WaitForPersistentTaskStatusFuture<Params extends PersistentTaskParams>
public static class WaitForPersistentTaskFuture<Params extends PersistentTaskParams>
extends PlainActionFuture<PersistentTask<Params>>
implements WaitForPersistentTaskStatusListener<Params> {
implements WaitForPersistentTaskListener<Params> {
}

public void testPersistentActionFailure() throws Exception {
PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
persistentTasksService.startPersistentTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
persistentTasksService.sendStartRequest(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
long allocationId = future.get().getAllocationId();
assertBusy(() -> {
// Wait for the task to start
Expand Down Expand Up @@ -108,7 +108,7 @@ public void testPersistentActionCompletion() throws Exception {
PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
String taskId = UUIDs.base64UUID();
persistentTasksService.startPersistentTask(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
persistentTasksService.sendStartRequest(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
long allocationId = future.get().getAllocationId();
assertBusy(() -> {
// Wait for the task to start
Expand All @@ -127,7 +127,7 @@ public void testPersistentActionCompletion() throws Exception {
logger.info("Simulating errant completion notification");
//try sending completion request with incorrect allocation id
PlainActionFuture<PersistentTask<?>> failedCompletionNotificationFuture = new PlainActionFuture<>();
persistentTasksService.sendCompletionNotification(taskId, Long.MAX_VALUE, null, failedCompletionNotificationFuture);
persistentTasksService.sendCompletionRequest(taskId, Long.MAX_VALUE, null, failedCompletionNotificationFuture);
assertThrows(failedCompletionNotificationFuture, ResourceNotFoundException.class);
// Make sure that the task is still running
assertThat(client().admin().cluster().prepareListTasks().setActions(TestPersistentTasksExecutor.NAME + "[c]")
Expand All @@ -142,7 +142,7 @@ public void testPersistentActionWithNoAvailableNode() throws Exception {
PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
TestParams testParams = new TestParams("Blah");
testParams.setExecutorNodeAttr("test");
persistentTasksService.startPersistentTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, testParams, future);
persistentTasksService.sendStartRequest(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, testParams, future);
String taskId = future.get().getId();

Settings nodeSettings = Settings.builder().put(nodeSettings(0)).put("node.attr.test_attr", "test").build();
Expand All @@ -169,14 +169,14 @@ public void testPersistentActionWithNoAvailableNode() throws Exception {

// Remove the persistent task
PlainActionFuture<PersistentTask<?>> removeFuture = new PlainActionFuture<>();
persistentTasksService.cancelPersistentTask(taskId, removeFuture);
persistentTasksService.sendRemoveRequest(taskId, removeFuture);
assertEquals(removeFuture.get().getId(), taskId);
}

public void testPersistentActionStatusUpdate() throws Exception {
PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
persistentTasksService.startPersistentTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
persistentTasksService.sendStartRequest(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
String taskId = future.get().getId();

assertBusy(() -> {
Expand All @@ -200,16 +200,16 @@ public void testPersistentActionStatusUpdate() throws Exception {
.get().getTasks().size(), equalTo(1));

int finalI = i;
WaitForPersistentTaskStatusFuture<?> future1 = new WaitForPersistentTaskStatusFuture<>();
persistentTasksService.waitForPersistentTaskStatus(taskId,
WaitForPersistentTaskFuture<?> future1 = new WaitForPersistentTaskFuture<>();
persistentTasksService.waitForPersistentTaskCondition(taskId,
task -> task != null && task.getStatus() != null && task.getStatus().toString() != null &&
task.getStatus().toString().equals("{\"phase\":\"phase " + (finalI + 1) + "\"}"),
TimeValue.timeValueSeconds(10), future1);
assertThat(future1.get().getId(), equalTo(taskId));
}

WaitForPersistentTaskStatusFuture<?> future1 = new WaitForPersistentTaskStatusFuture<>();
persistentTasksService.waitForPersistentTaskStatus(taskId,
WaitForPersistentTaskFuture<?> future1 = new WaitForPersistentTaskFuture<>();
persistentTasksService.waitForPersistentTaskCondition(taskId,
task -> false, TimeValue.timeValueMillis(10), future1);

assertThrows(future1, IllegalStateException.class, "timed out after 10ms");
Expand All @@ -220,8 +220,8 @@ public void testPersistentActionStatusUpdate() throws Exception {
" and allocation id -2 doesn't exist");

// Wait for the task to disappear
WaitForPersistentTaskStatusFuture<?> future2 = new WaitForPersistentTaskStatusFuture<>();
persistentTasksService.waitForPersistentTaskStatus(taskId, Objects::isNull, TimeValue.timeValueSeconds(10), future2);
WaitForPersistentTaskFuture<?> future2 = new WaitForPersistentTaskFuture<>();
persistentTasksService.waitForPersistentTaskCondition(taskId, Objects::isNull, TimeValue.timeValueSeconds(10), future2);

logger.info("Completing the running task");
// Complete the running task and make sure it finishes properly
Expand All @@ -235,11 +235,11 @@ public void testCreatePersistentTaskWithDuplicateId() throws Exception {
PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
String taskId = UUIDs.base64UUID();
persistentTasksService.startPersistentTask(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
persistentTasksService.sendStartRequest(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
future.get();

PlainActionFuture<PersistentTask<TestParams>> future2 = new PlainActionFuture<>();
persistentTasksService.startPersistentTask(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future2);
persistentTasksService.sendStartRequest(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future2);
assertThrows(future2, ResourceAlreadyExistsException.class);

assertBusy(() -> {
Expand Down
6 changes: 3 additions & 3 deletions server/src/test/java/org/elasticsearch/persistent/PersistentTasksNodeServiceTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -235,14 +235,14 @@ public void testTaskCancellation() {
AtomicReference<ActionListener<CancelTasksResponse>> capturedListener = new AtomicReference<>();
PersistentTasksService persistentTasksService = new PersistentTasksService(Settings.EMPTY, null, null, null) {
@Override
public void sendTaskManagerCancellation(long taskId, ActionListener<CancelTasksResponse> listener) {
void sendCancelRequest(final long taskId, final String reason, final ActionListener<CancelTasksResponse> listener) {
capturedTaskId.set(taskId);
capturedListener.set(listener);
}

@Override
public void sendCompletionNotification(String taskId, long allocationId, Exception failure,
ActionListener<PersistentTask<?>> listener) {
public void sendCompletionRequest(final String taskId, final long taskAllocationId,
final Exception taskFailure, final ActionListener<PersistentTask<?>> listener) {
fail("Shouldn't be called during Cluster State cancellation");
}
};
Expand Down
2 changes: 1 addition & 1 deletion server/src/test/java/org/elasticsearch/persistent/decider/EnableAssignmentDeciderIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ public void testEnableAssignmentAfterRestart() throws Exception {
final CountDownLatch latch = new CountDownLatch(numberOfTasks);
for (int i = 0; i < numberOfTasks; i++) {
PersistentTasksService service = internalCluster().getInstance(PersistentTasksService.class);
service.startPersistentTask("task_" + i, TestPersistentTasksExecutor.NAME, randomTaskParams(),
service.sendStartRequest("task_" + i, TestPersistentTasksExecutor.NAME, randomTaskParams(),
new ActionListener<PersistentTask<PersistentTaskParams>>() {
@Override
public void onResponse(PersistentTask<PersistentTaskParams> task) {
Expand Down
4 changes: 2 additions & 2 deletions ...ck/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportCloseJobAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ private void forceCloseJob(ClusterState currentState, CloseJobAction.Request req
PersistentTasksCustomMetaData.PersistentTask<?> jobTask = MlMetadata.getJobTask(jobId, tasks);
if (jobTask != null) {
auditor.info(jobId, Messages.JOB_AUDIT_FORCE_CLOSING);
persistentTasksService.cancelPersistentTask(jobTask.getId(),
persistentTasksService.sendRemoveRequest(jobTask.getId(),
new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {
Expand Down Expand Up @@ -400,7 +400,7 @@ public boolean hasJobsToWaitFor() {
// so wait for that to happen here.
void waitForJobClosed(CloseJobAction.Request request, WaitForCloseRequest waitForCloseRequest, CloseJobAction.Response response,
ActionListener<CloseJobAction.Response> listener) {
persistentTasksService.waitForPersistentTasksStatus(persistentTasksCustomMetaData -> {
persistentTasksService.waitForPersistentTasksCondition(persistentTasksCustomMetaData -> {
for (String persistentTaskId : waitForCloseRequest.persistentTaskIds) {
if (persistentTasksCustomMetaData.getTask(persistentTaskId) != null) {
return false;
Expand Down
2 changes: 1 addition & 1 deletion ...gin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDatafeedAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,7 @@ private void removeDatafeedTask(DeleteDatafeedAction.Request request, ClusterSta
if (datafeedTask == null) {
listener.onResponse(true);
} else {
persistentTasksService.cancelPersistentTask(datafeedTask.getId(),
persistentTasksService.sendRemoveRequest(datafeedTask.getId(),
new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
Expand Down
2 changes: 1 addition & 1 deletion ...k/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteJobAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,7 @@ private void removePersistentTask(String jobId, ClusterState currentState,
if (jobTask == null) {
listener.onResponse(null);
} else {
persistentTasksService.cancelPersistentTask(jobTask.getId(),
persistentTasksService.sendRemoveRequest(jobTask.getId(),
new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {
Expand Down
8 changes: 4 additions & 4 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportOpenJobAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -449,7 +449,7 @@ public void onFailure(Exception e) {

// Step 4. Start job task
ActionListener<PutJobAction.Response> jobUpateListener = ActionListener.wrap(
response -> persistentTasksService.startPersistentTask(MlMetadata.jobTaskId(jobParams.getJobId()),
response -> persistentTasksService.sendStartRequest(MlMetadata.jobTaskId(jobParams.getJobId()),
OpenJobAction.TASK_NAME, jobParams, finalListener),
listener::onFailure
);
Expand Down Expand Up @@ -523,8 +523,8 @@ public void onFailure(Exception e) {

private void waitForJobStarted(String taskId, OpenJobAction.JobParams jobParams, ActionListener<OpenJobAction.Response> listener) {
JobPredicate predicate = new JobPredicate();
persistentTasksService.waitForPersistentTaskStatus(taskId, predicate, jobParams.getTimeout(),
new PersistentTasksService.WaitForPersistentTaskStatusListener<OpenJobAction.JobParams>() {
persistentTasksService.waitForPersistentTaskCondition(taskId, predicate, jobParams.getTimeout(),
new PersistentTasksService.WaitForPersistentTaskListener<OpenJobAction.JobParams>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<OpenJobAction.JobParams> persistentTask) {
if (predicate.exception != null) {
Expand Down Expand Up @@ -555,7 +555,7 @@ public void onTimeout(TimeValue timeout) {

private void cancelJobStart(PersistentTasksCustomMetaData.PersistentTask<OpenJobAction.JobParams> persistentTask, Exception exception,
ActionListener<OpenJobAction.Response> listener) {
persistentTasksService.cancelPersistentTask(persistentTask.getId(),
persistentTasksService.sendRemoveRequest(persistentTask.getId(),
new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {
Expand Down
Loading

0 comments on commit 02b6ea3

Please sign in to comment.