Only start re-assigning persistent tasks if they are not already being reassigned

server/src/main/java/org/elasticsearch/persistent/PersistentTasksClusterService.java

@@ -37,6 +37,7 @@
 import java.io.Closeable;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 /**
@@ -55,6 +56,7 @@ public class PersistentTasksClusterService implements ClusterStateListener, Clos
     private final EnableAssignmentDecider enableDecider;
     private final ThreadPool threadPool;
     private final PeriodicRechecker periodicRechecker;
+    private final AtomicBoolean reassigningTasks = new AtomicBoolean(false);
 
     public PersistentTasksClusterService(Settings settings, PersistentTasksExecutorRegistry registry, ClusterService clusterService,
                                          ThreadPool threadPool) {
@@ -353,7 +355,10 @@ public void clusterChanged(ClusterChangedEvent event) {
     /**
      * Submit a cluster state update to reassign any persistent tasks that need reassigning
      */
-    private void reassignPersistentTasks() {
+    void reassignPersistentTasks() {
+        if (this.reassigningTasks.compareAndSet(false, true) == false) {
+            return;
+        }
         clusterService.submitStateUpdateTask("reassign persistent tasks", new ClusterStateUpdateTask() {
             @Override
             public ClusterState execute(ClusterState currentState) {
@@ -362,6 +367,7 @@ public ClusterState execute(ClusterState currentState) {
 
             @Override
             public void onFailure(String source, Exception e) {
+                reassigningTasks.set(false);
                 logger.warn("failed to reassign persistent tasks", e);
                 if (e instanceof NotMasterException == false) {
                     // There must be a task that's worth rechecking because there was one
@@ -373,6 +379,7 @@ public void onFailure(String source, Exception e) {
 
             @Override
             public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+                reassigningTasks.set(false);
                 if (isAnyTaskUnassigned(newState.getMetadata().custom(PersistentTasksCustomMetadata.TYPE))) {
                     periodicRechecker.rescheduleIfNecessary();
                 }

server/src/test/java/org/elasticsearch/persistent/PersistentTasksClusterServiceTests.java

@@ -50,6 +50,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
@@ -69,8 +70,13 @@
 import static org.hamcrest.core.Is.is;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
 public class PersistentTasksClusterServiceTests extends ESTestCase {
@@ -429,7 +435,7 @@ public void testPeriodicRecheck() throws Exception {
         nonClusterStateCondition = false;
 
         boolean shouldSimulateFailure = randomBoolean();
-        ClusterService recheckTestClusterService = createRecheckTestClusterService(clusterState, shouldSimulateFailure);
+        ClusterService recheckTestClusterService = createStateUpdateClusterState(clusterState, shouldSimulateFailure);
         PersistentTasksClusterService service = createService(recheckTestClusterService,
             (params, candidateNodes, currentState) -> assignBasedOnNonClusterStateCondition(candidateNodes));
 
@@ -477,7 +483,7 @@ public void testPeriodicRecheckOffMaster() {
         ClusterState clusterState = builder.metadata(metadata).nodes(nodes).build();
         nonClusterStateCondition = false;
 
-        ClusterService recheckTestClusterService = createRecheckTestClusterService(clusterState, false);
+        ClusterService recheckTestClusterService = createStateUpdateClusterState(clusterState, false);
         PersistentTasksClusterService service = createService(recheckTestClusterService,
             (params, candidateNodes, currentState) -> assignBasedOnNonClusterStateCondition(candidateNodes));
 
@@ -639,7 +645,49 @@ public void testTasksNotAssignedToShuttingDownNodes() {
             Sets.haveEmptyIntersection(shutdownNodes, nodesWithTasks));
     }
 
-    private ClusterService createRecheckTestClusterService(ClusterState initialState, boolean shouldSimulateFailure) {
+    public void testReassignOnlyOnce() throws Exception {
+        CountDownLatch latch = new CountDownLatch(1);
+        ClusterState initialState = initialState();
+        ClusterState.Builder builder = ClusterState.builder(initialState);
+        PersistentTasksCustomMetadata.Builder tasks = PersistentTasksCustomMetadata.builder(
+            initialState.metadata().custom(PersistentTasksCustomMetadata.TYPE)
+        );
+        DiscoveryNodes.Builder nodes = DiscoveryNodes.builder(initialState.nodes());
+        addTestNodes(nodes, randomIntBetween(1, 3));
+        addTask(tasks, "assign_based_on_non_cluster_state_condition", null);
+        Metadata.Builder metadata = Metadata.builder(initialState.metadata()).putCustom(PersistentTasksCustomMetadata.TYPE, tasks.build());
+        ClusterState clusterState = builder.metadata(metadata).nodes(nodes).build();
+
+        boolean shouldSimulateFailure = randomBoolean();
+        ClusterService recheckTestClusterService = createStateUpdateClusterState(clusterState, shouldSimulateFailure, latch);
+        PersistentTasksClusterService service = createService(
+            recheckTestClusterService,
+            (params, candidateNodes, currentState) -> assignBasedOnNonClusterStateCondition(candidateNodes)
+        );
+        verify(recheckTestClusterService, atLeastOnce()).getClusterSettings();
+        verify(recheckTestClusterService, atLeastOnce()).addListener(any());
+        Thread t1 = new Thread(service::reassignPersistentTasks);
+        Thread t2 = new Thread(service::reassignPersistentTasks);
+        try {
+            t1.start();
+            t2.start();
+            assertBusy(() -> verify(recheckTestClusterService, times(1)).submitStateUpdateTask(eq("reassign persistent tasks"), any()));
+            latch.countDown();
+            t1.join();
+            t2.join();
+            verifyNoMoreInteractions(recheckTestClusterService);
+        } finally {
+            latch.countDown();
+            t1.join();
+            t2.join();
+        }
+    }
+
+    private ClusterService createStateUpdateClusterState(ClusterState initialState, boolean shouldSimulateFailure) {
+        return createStateUpdateClusterState(initialState, shouldSimulateFailure, null);
+    }
+
+    private ClusterService createStateUpdateClusterState(ClusterState initialState, boolean shouldSimulateFailure, CountDownLatch await) {
         AtomicBoolean testFailureNextTime = new AtomicBoolean(shouldSimulateFailure);
         AtomicReference<ClusterState> state = new AtomicReference<>(initialState);
         ClusterService recheckTestClusterService = mock(ClusterService.class);
@@ -651,6 +699,9 @@ private ClusterService createRecheckTestClusterService(ClusterState initialState
             ClusterStateUpdateTask task = (ClusterStateUpdateTask) invocationOnMock.getArguments()[1];
             ClusterState before = state.get();
             ClusterState after = task.execute(before);
+            if (await != null) {
+                await.await();
+            }
             if (testFailureNextTime.compareAndSet(true, false)) {
                 task.onFailure("testing failure", new RuntimeException("foo"));
             } else {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobNodeSelector.java

@@ -23,11 +23,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
+import java.util.TreeMap;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -156,7 +156,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
             logger.debug(reason);
             return new PersistentTasksCustomMetadata.Assignment(null, reason);
         }
-        List<String> reasons = new LinkedList<>();
+        Map<String, String> reasons = new TreeMap<>();
         long maxAvailableMemory = Long.MIN_VALUE;
         DiscoveryNode minLoadedNodeByMemory = null;
         long requiredMemoryForJob = estimatedMemoryFootprint;
@@ -166,7 +166,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
             String reason = nodeFilter.apply(node);
             if (reason != null) {
                 logger.trace(reason);
-                reasons.add(reason);
+                reasons.put(node.getName(), reason);
                 continue;
             }
             NodeLoad currentLoad = nodeLoadDetector.detectNodeLoad(
@@ -180,7 +180,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
             if (currentLoad.getError() != null) {
                 reason = createReason(jobId, nodeNameAndMlAttributes(node), currentLoad.getError());
                 logger.trace(reason);
-                reasons.add(reason);
+                reasons.put(node.getName(), reason);
                 continue;
             }
             // Assuming the node is eligible at all, check loading
@@ -194,7 +194,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
                     currentLoad.getNumAllocatingJobs(),
                     maxConcurrentJobAllocations);
                 logger.trace(reason);
-                reasons.add(reason);
+                reasons.put(node.getName(), reason);
                 continue;
             }
 
@@ -206,7 +206,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
                     MAX_OPEN_JOBS_PER_NODE.getKey(),
                     maxNumberOfOpenJobs);
                 logger.trace(reason);
-                reasons.add(reason);
+                reasons.put(node.getName(), reason);
                 continue;
             }
 
@@ -215,7 +215,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
                     nodeNameAndMlAttributes(node),
                     "This node is not providing accurate information to determine is load by memory.");
                 logger.trace(reason);
-                reasons.add(reason);
+                reasons.put(node.getName(),reason);
                 continue;
             }
 
@@ -224,7 +224,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
                     nodeNameAndMlAttributes(node),
                     "This node is indicating that it has no native memory for machine learning.");
                 logger.trace(reason);
-                reasons.add(reason);
+                reasons.put(node.getName(),reason);
                 continue;
             }
 
@@ -247,7 +247,7 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
                     requiredMemoryForJob,
                     ByteSizeValue.ofBytes(requiredMemoryForJob).toString());
                 logger.trace(reason);
-                reasons.add(reason);
+                reasons.put(node.getName(),reason);
                 continue;
             }
 
@@ -260,15 +260,15 @@ public PersistentTasksCustomMetadata.Assignment selectNode(Long estimatedMemoryF
         return createAssignment(
             estimatedMemoryFootprint,
             minLoadedNodeByMemory,
-            reasons,
+            reasons.values(),
             maxNodeSize > 0L ?
                 NativeMemoryCalculator.allowedBytesForMl(maxNodeSize, maxMachineMemoryPercent, useAutoMemoryPercentage) :
                 Long.MAX_VALUE);
     }
 
     PersistentTasksCustomMetadata.Assignment createAssignment(long estimatedMemoryUsage,
                                                               DiscoveryNode minLoadedNode,
-                                                              List<String> reasons,
+                                                              Collection<String> reasons,
                                                               long biggestPossibleJob) {
         if (minLoadedNode == null) {
             String explanation = String.join("|", reasons);