Merge pull request #15700 from jasontedor/serial-cluster-state-task-executions

Add test that cluster state update tasks are executed in order

Closes #15483

core/src/test/java/org/elasticsearch/cluster/ClusterServiceIT.java

@@ -27,6 +27,7 @@ import org.elasticsearch.cluster.service.InternalClusterService;
 import org.elasticsearch.cluster.service.PendingClusterTask;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.component.LifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
@@ -51,9 +52,12 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -61,6 +65,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.elasticsearch.common.settings.Settings.settingsBuilder;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -796,6 +801,91 @@ public class ClusterServiceIT extends ESIntegTestCase {
         assertTrue(published.get());
     }
 
+    // test that for a single thread, tasks are executed in the order
+    // that they are submitted
+    public void testClusterStateUpdateTasksAreExecutedInOrder() throws BrokenBarrierException, InterruptedException {
+        Settings settings = settingsBuilder()
+            .put("discovery.type", "local")
+            .build();
+        internalCluster().startNode(settings);
+        ClusterService clusterService = internalCluster().getInstance(ClusterService.class);
+
+        class TaskExecutor implements ClusterStateTaskExecutor<Integer> {
+            List<Integer> tasks = new ArrayList<>();
+
+            @Override
+            public BatchResult<Integer> execute(ClusterState currentState, List<Integer> tasks) throws Exception {
+                this.tasks.addAll(tasks);
+                return BatchResult.<Integer>builder().successes(tasks).build(ClusterState.builder(currentState).build());
+            }
+
+            @Override
+            public boolean runOnlyOnMaster() {
+                return false;
+            }
+        }
+
+        int numberOfThreads = randomIntBetween(2, 8);
+        TaskExecutor[] executors = new TaskExecutor[numberOfThreads];
+        for (int i = 0; i < numberOfThreads; i++) {
+            executors[i] = new TaskExecutor();
+        }
+
+        int tasksSubmittedPerThread = randomIntBetween(2, 1024);
+
+        CopyOnWriteArrayList<Tuple<String, Throwable>> failures = new CopyOnWriteArrayList<>();
+        CountDownLatch updateLatch = new CountDownLatch(numberOfThreads * tasksSubmittedPerThread);
+
+        ClusterStateTaskListener listener = new ClusterStateTaskListener() {
+            @Override
+            public void onFailure(String source, Throwable t) {
+                logger.error("unexpected failure: [{}]", t, source);
+                failures.add(new Tuple<>(source, t));
+                updateLatch.countDown();
+            }
+
+            @Override
+            public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+                updateLatch.countDown();
+            }
+        };
+
+        CyclicBarrier barrier = new CyclicBarrier(1 + numberOfThreads);
+
+        for (int i = 0; i < numberOfThreads; i++) {
+            final int index = i;
+            Thread thread = new Thread(() -> {
+                try {
+                    barrier.await();
+                    for (int j = 0; j < tasksSubmittedPerThread; j++) {
+                        clusterService.submitStateUpdateTask("[" + index + "][" + j + "]", j, ClusterStateTaskConfig.build(randomFrom(Priority.values())), executors[index], listener);
+                    }
+                    barrier.await();
+                } catch (InterruptedException | BrokenBarrierException e) {
+                    throw new AssertionError(e);
+                }
+            });
+            thread.start();
+        }
+
+        // wait for all threads to be ready
+        barrier.await();
+        // wait for all threads to finish
+        barrier.await();
+
+        updateLatch.await();
+
+        assertThat(failures, empty());
+
+        for (int i = 0; i < numberOfThreads; i++) {
+            assertEquals(tasksSubmittedPerThread, executors[i].tasks.size());
+            for (int j = 0; j < tasksSubmittedPerThread; j++) {
+                assertNotNull(executors[i].tasks.get(j));
+                assertEquals("cluster state update task executed out of order", j, (int)executors[i].tasks.get(j));
+            }
+        }
+    }
+
     public void testClusterStateBatchedUpdates() throws InterruptedException {
         Settings settings = settingsBuilder()
                 .put("discovery.type", "local")