Merge pull request #14993 from jasontedor/tribe-node-cluster-state-batch

Tribe nodes should apply cluster state updates in batches

core/src/main/java/org/elasticsearch/tribe/TribeService.java

@@ -26,7 +26,8 @@ import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
-import org.elasticsearch.cluster.ClusterStateUpdateTask;
+import org.elasticsearch.cluster.ClusterStateTaskConfig;
+import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.block.ClusterBlocks;
@@ -36,6 +37,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.RoutingTable;
+import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
@@ -205,26 +207,61 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
         }
     }
 
-    class TribeClusterStateListener implements ClusterStateListener {
 
+    class TribeClusterStateListener implements ClusterStateListener {
         private final String tribeName;
+        private final TribeNodeClusterStateTaskExecutor executor;
 
         TribeClusterStateListener(Node tribeNode) {
-            this.tribeName = tribeNode.settings().get(TRIBE_NAME);
+            String tribeName = tribeNode.settings().get(TRIBE_NAME);
+            this.tribeName = tribeName;
+            executor = new TribeNodeClusterStateTaskExecutor(tribeName);
         }
 
         @Override
         public void clusterChanged(final ClusterChangedEvent event) {
             logger.debug("[{}] received cluster event, [{}]", tribeName, event.source());
-            clusterService.submitStateUpdateTask("cluster event from " + tribeName + ", " + event.source(), new ClusterStateUpdateTask() {
+            clusterService.submitStateUpdateTask(
+                    "cluster event from " + tribeName + ", " + event.source(),
+                    event,
+                    ClusterStateTaskConfig.build(Priority.NORMAL),
+                    executor,
+                    (source, t) -> logger.warn("failed to process [{}]", t, source));
+        }
+    }
+
+    class TribeNodeClusterStateTaskExecutor implements ClusterStateTaskExecutor<ClusterChangedEvent> {
+        private final String tribeName;
+
+        TribeNodeClusterStateTaskExecutor(String tribeName) {
+            this.tribeName = tribeName;
+        }
+
+
         @Override
         public boolean runOnlyOnMaster() {
             return false;
         }
 
         @Override
-                public ClusterState execute(ClusterState currentState) throws Exception {
+        public BatchResult<ClusterChangedEvent> execute(ClusterState currentState, List<ClusterChangedEvent> tasks) throws Exception {
-                    ClusterState tribeState = event.state();
+            ClusterState accumulator = ClusterState.builder(currentState).build();
+            BatchResult.Builder<ClusterChangedEvent> builder = BatchResult.builder();
+
+            try {
+                // we only need to apply the latest cluster state update
+                accumulator = applyUpdate(accumulator, tasks.get(tasks.size() - 1));
+                builder.successes(tasks);
+            } catch (Throwable t) {
+                builder.failures(tasks, t);
+            }
+
+            return builder.build(accumulator);
+        }
+
+        private ClusterState applyUpdate(ClusterState currentState, ClusterChangedEvent task) {
+            boolean clusterStateChanged = false;
+            ClusterState tribeState = task.state();
             DiscoveryNodes.Builder nodes = DiscoveryNodes.builder(currentState.nodes());
             // -- merge nodes
             // go over existing nodes, and see if they need to be removed
@@ -232,6 +269,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                 String markedTribeName = discoNode.attributes().get(TRIBE_NAME);
                 if (markedTribeName != null && markedTribeName.equals(tribeName)) {
                     if (tribeState.nodes().get(discoNode.id()) == null) {
+                        clusterStateChanged = true;
                         logger.info("[{}] removing node [{}]", tribeName, discoNode);
                         nodes.remove(discoNode.id());
                     }
@@ -247,6 +285,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                     }
                     tribeAttr.put(TRIBE_NAME, tribeName);
                     DiscoveryNode discoNode = new DiscoveryNode(tribe.name(), tribe.id(), tribe.getHostName(), tribe.getHostAddress(), tribe.address(), unmodifiableMap(tribeAttr), tribe.version());
+                    clusterStateChanged = true;
                     logger.info("[{}] adding node [{}]", tribeName, discoNode);
                     nodes.put(discoNode);
                 }
@@ -261,6 +300,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                 String markedTribeName = index.getSettings().get(TRIBE_NAME);
                 if (markedTribeName != null && markedTribeName.equals(tribeName)) {
                     IndexMetaData tribeIndex = tribeState.metaData().index(index.getIndex());
+                    clusterStateChanged = true;
                     if (tribeIndex == null || tribeIndex.getState() == IndexMetaData.State.CLOSE) {
                         logger.info("[{}] removing index [{}]", tribeName, index.getIndex());
                         removeIndex(blocks, metaData, routingTable, index);
@@ -284,6 +324,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                 if (indexMetaData == null) {
                     if (!droppedIndices.contains(tribeIndex.getIndex())) {
                         // a new index, add it, and add the tribe name as a setting
+                        clusterStateChanged = true;
                         logger.info("[{}] adding index [{}]", tribeName, tribeIndex.getIndex());
                         addNewIndex(tribeState, blocks, metaData, routingTable, tribeIndex);
                     }
@@ -295,6 +336,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                             // we chose any tribe, carry on
                         } else if (ON_CONFLICT_DROP.equals(onConflict)) {
                             // drop the indices, there is a conflict
+                            clusterStateChanged = true;
                             logger.info("[{}] dropping index [{}] due to conflict with [{}]", tribeName, tribeIndex.getIndex(), existingFromTribe);
                             removeIndex(blocks, metaData, routingTable, tribeIndex);
                             droppedIndices.add(tribeIndex.getIndex());
@@ -303,6 +345,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                             String preferredTribeName = onConflict.substring(ON_CONFLICT_PREFER.length());
                             if (tribeName.equals(preferredTribeName)) {
                                 // the new one is hte preferred one, replace...
+                                clusterStateChanged = true;
                                 logger.info("[{}] adding index [{}], preferred over [{}]", tribeName, tribeIndex.getIndex(), existingFromTribe);
                                 removeIndex(blocks, metaData, routingTable, tribeIndex);
                                 addNewIndex(tribeState, blocks, metaData, routingTable, tribeIndex);
@@ -312,8 +355,12 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                 }
             }
 
+            if (!clusterStateChanged) {
+                return currentState;
+            } else {
                 return ClusterState.builder(currentState).incrementVersion().blocks(blocks).nodes(nodes).metaData(metaData).routingTable(routingTable.build()).build();
             }
+        }
 
         private void removeIndex(ClusterBlocks.Builder blocks, MetaData.Builder metaData, RoutingTable.Builder routingTable, IndexMetaData index) {
             metaData.remove(index.getIndex());
@@ -335,12 +382,5 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
                 blocks.addIndexBlock(tribeIndex.getIndex(), IndexMetaData.INDEX_WRITE_BLOCK);
             }
         }
-
-                @Override
-                public void onFailure(String source, Throwable t) {
-                    logger.warn("failed to process [{}]", t, source);
-                }
-            });
-        }
     }
 }