Persistent Tasks: check the current state in waitForPersistentTaskStatus (elastic/x-pack-elasticsearch#935)
Add a check for the current state waitForPersistentTaskStatus before waiting for the next one. This fixes sporadic failure in testPersistentActionStatusUpdate test.
relates elastic/x-pack-elasticsearch#928
Original commit: elastic/x-pack-elasticsearch@0db4ac92d275ec3a0ec203a9466e641f33ef50b6
@@ -117,11 +117,15 @@ public class PersistentTasksService extends AbstractComponent {
}
/**
- * Waits for the persistent task with giving id (taskId) to achieve the desired status.
+ * Checks if the persistent task with giving id (taskId) has the desired state and if it doesn't
+ * waits of it.
*/
public void waitForPersistentTaskStatus(long taskId, Predicate<PersistentTask<?>> predicate, @Nullable TimeValue timeout,
WaitForPersistentTaskStatusListener listener) {
ClusterStateObserver stateObserver = new ClusterStateObserver(clusterService, timeout, logger, threadPool.getThreadContext());
+ if (predicate.test(PersistentTasksCustomMetaData.getTaskWithId(stateObserver.setAndGetObservedState(), taskId))) {
+ listener.onResponse(taskId);
+ } else {
stateObserver.waitForNextChange(new ClusterStateObserver.Listener() {
@Override
public void onNewClusterState(ClusterState state) {
@@ -131,7 +135,6 @@ public class PersistentTasksService extends AbstractComponent {
@Override
public void onClusterServiceClose() {
listener.onFailure(new NodeClosedException(clusterService.localNode()));
-
}
@Override
@@ -140,6 +143,7 @@ public class PersistentTasksService extends AbstractComponent {
}
}, clusterState -> predicate.test(PersistentTasksCustomMetaData.getTaskWithId(clusterState, taskId)));
}
+ }
public interface WaitForPersistentTaskStatusListener extends PersistentTaskOperationListener {
default void onTimeout(TimeValue timeout) {
