Skip to content
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

Jump to bottom

Rename methods in PersistentTasksService #30837

Merged
merged 3 commits into from
May 30, 2018
Merged

Rename methods in PersistentTasksService #30837

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
57f6196
Rename methods in PersistentTasksService
tlrx May 23, 2018
b4a50b7
Revert sendUpdateStateRequest -> updateStatus
tlrx May 29, 2018
ab05475
waitForPersistentTaskCondition
tlrx May 29, 2018
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
16 changes: 10 additions & 6 deletions server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop

Large diffs are not rendered by default.

3 changes: 1 addition & 2 deletions server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorFullRestartIT.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -465,7 +465,7 @@ public void onFailure(Exception e) {

// Step 4. Start job task
ActionListener<PutJobAction.Response> establishedMemoryUpdateListener = 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 @@ -518,8 +518,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 @@ -550,7 +550,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