honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Persistent Tasks: check the current state in waitForPersistentTaskStatus (#935) 

Add a check for the current state waitForPersistentTaskStatus before waiting for the next one. This fixes sporadic failure in testPersistentActionStatusUpdate test.

Fixes #928
This commit is contained in:
Igor Motov 2017-04-04 09:44:56 -04:00 committed by Martijn van Groningen
parent a5acb556b0
commit 5b45b167bd
No known key found for this signature in database
GPG Key ID: AB236F4FCF2AF12A
1 changed files with 20 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java
View File

@ -130,11 +130,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, public void waitForPersistentTaskStatus(long taskId, Predicate<PersistentTask<?>> predicate, @Nullable TimeValue timeout,
WaitForPersistentTaskStatusListener listener) { WaitForPersistentTaskStatusListener listener) {
ClusterStateObserver stateObserver = new ClusterStateObserver(clusterService, timeout, logger, threadPool.getThreadContext()); 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() { stateObserver.waitForNextChange(new ClusterStateObserver.Listener() {
@Override @Override
public void onNewClusterState(ClusterState state) { public void onNewClusterState(ClusterState state) {
@ -144,7 +148,6 @@ public class PersistentTasksService extends AbstractComponent {
@Override @Override
public void onClusterServiceClose() { public void onClusterServiceClose() {
listener.onFailure(new NodeClosedException(clusterService.localNode())); listener.onFailure(new NodeClosedException(clusterService.localNode()));
} }
@Override @Override
@ -153,6 +156,7 @@ public class PersistentTasksService extends AbstractComponent {
} }
}, clusterState -> predicate.test(PersistentTasksCustomMetaData.getTaskWithId(clusterState, taskId))); }, clusterState -> predicate.test(PersistentTasksCustomMetaData.getTaskWithId(clusterState, taskId)));
} }
}
public interface WaitForPersistentTaskStatusListener extends PersistentTaskOperationListener { public interface WaitForPersistentTaskStatusListener extends PersistentTaskOperationListener {
default void onTimeout(TimeValue timeout) { default void onTimeout(TimeValue timeout) {