Automatic index creation can still cause "index missing" failures, closes #1199.

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/ClusterChangedEvent.java

@@ -20,6 +20,10 @@
 package org.elasticsearch.cluster;
 
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.collect.ImmutableList;
+import org.elasticsearch.common.collect.Lists;
+
+import java.util.List;
 
 /**
  * @author kimchy (shay.banon)
@@ -70,6 +74,50 @@ public class ClusterChangedEvent {
         return true;
     }
 
+    /**
+     * Returns the indices created in this event
+     */
+    public List<String> indicesCreated() {
+        if (previousState == null) {
+            return Lists.newArrayList(state.metaData().indices().keySet());
+        }
+        if (!metaDataChanged()) {
+            return ImmutableList.of();
+        }
+        List<String> created = null;
+        for (String index : state.metaData().indices().keySet()) {
+            if (!previousState.metaData().hasIndex(index)) {
+                if (created == null) {
+                    created = Lists.newArrayList();
+                }
+                created.add(index);
+            }
+        }
+        return created == null ? ImmutableList.<String>of() : created;
+    }
+
+    /**
+     * Returns the indices deleted in this event
+     */
+    public List<String> indicesDeleted() {
+        if (previousState == null) {
+            return ImmutableList.of();
+        }
+        if (!metaDataChanged()) {
+            return ImmutableList.of();
+        }
+        List<String> deleted = null;
+        for (String index : previousState.metaData().indices().keySet()) {
+            if (!state.metaData().hasIndex(index)) {
+                if (deleted == null) {
+                    deleted = Lists.newArrayList();
+                }
+                deleted.add(index);
+            }
+        }
+        return deleted == null ? ImmutableList.<String>of() : deleted;
+    }
+
     public boolean metaDataChanged() {
         return state.metaData() != previousState.metaData();
     }

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/metadata/MetaDataCreateIndexService.java

@@ -27,9 +27,7 @@ import org.elasticsearch.cluster.action.index.NodeIndexCreatedAction;
 import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
-import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RoutingTable;
-import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.ShardsAllocation;
 import org.elasticsearch.common.Strings;
@@ -63,7 +61,11 @@ import org.elasticsearch.threadpool.ThreadPool;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
-import java.util.*;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -118,8 +120,6 @@ public class MetaDataCreateIndexService extends AbstractComponent {
 
 
         clusterService.submitStateUpdateTask("create-index [" + request.index + "], cause [" + request.cause + "]", new ProcessedClusterStateUpdateTask() {
-            final Set<String> allocatedNodes = Sets.newHashSet();
-
             @Override public ClusterState execute(ClusterState currentState) {
                 try {
                     try {
@@ -271,43 +271,28 @@ public class MetaDataCreateIndexService extends AbstractComponent {
                         updatedState = newClusterStateBuilder().state(updatedState).routingResult(routingResult).build();
                     }
 
-                    // initialize the counter only for nodes the shards are allocated to
+                    // we wait for events from all nodes that the index has been added to the metadata
-                    if (updatedState.routingTable().hasIndex(request.index)) {
+                    final AtomicInteger counter = new AtomicInteger(currentState.nodes().size());
-                        for (IndexShardRoutingTable indexShardRoutingTable : updatedState.routingTable().index(request.index)) {
+
-                            for (ShardRouting shardRouting : indexShardRoutingTable) {
+                    final NodeIndexCreatedAction.Listener nodeIndexCreatedListener = new NodeIndexCreatedAction.Listener() {
-                                // if we have a routing for this shard on a node, and its not the master node (since we already created
+                        @Override public void onNodeIndexCreated(String index, String nodeId) {
-                                // an index on it), then add it
+                            if (index.equals(request.index)) {
-                                if (shardRouting.currentNodeId() != null && !updatedState.nodes().localNodeId().equals(shardRouting.currentNodeId())) {
+                                if (counter.decrementAndGet() == 0) {
-                                    allocatedNodes.add(shardRouting.currentNodeId());
+                                    listener.onResponse(new Response(true, indexMetaData));
+                                    nodeIndexCreatedAction.remove(this);
                                 }
                             }
                         }
-                    }
+                    };
 
-                    if (!allocatedNodes.isEmpty()) {
+                    nodeIndexCreatedAction.add(nodeIndexCreatedListener);
-                        final AtomicInteger counter = new AtomicInteger(allocatedNodes.size());
-
-                        final NodeIndexCreatedAction.Listener nodeIndexCreatedListener = new NodeIndexCreatedAction.Listener() {
-                            @Override public void onNodeIndexCreated(String index, String nodeId) {
-                                if (index.equals(request.index)) {
-                                    if (counter.decrementAndGet() == 0) {
-                                        listener.onResponse(new Response(true, indexMetaData));
-                                        nodeIndexCreatedAction.remove(this);
-                                    }
-                                }
-                            }
-                        };
-
-                        nodeIndexCreatedAction.add(nodeIndexCreatedListener);
-
-                        listener.future = threadPool.schedule(request.timeout, ThreadPool.Names.SAME, new Runnable() {
-                            @Override public void run() {
-                                listener.onResponse(new Response(false, indexMetaData));
-                                nodeIndexCreatedAction.remove(nodeIndexCreatedListener);
-                            }
-                        });
-                    }
 
+                    listener.future = threadPool.schedule(request.timeout, ThreadPool.Names.SAME, new Runnable() {
+                        @Override public void run() {
+                            listener.onResponse(new Response(false, indexMetaData));
+                            nodeIndexCreatedAction.remove(nodeIndexCreatedListener);
+                        }
+                    });
 
                     return updatedState;
                 } catch (Exception e) {
@@ -318,9 +303,6 @@ public class MetaDataCreateIndexService extends AbstractComponent {
             }
 
             @Override public void clusterStateProcessed(ClusterState clusterState) {
-                if (allocatedNodes.isEmpty()) {
-                    listener.onResponse(new Response(true, clusterState.metaData().index(request.index)));
-                }
             }
         });
     }

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/metadata/MetaDataDeleteIndexService.java

@@ -24,12 +24,9 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.action.index.NodeIndexDeletedAction;
 import org.elasticsearch.cluster.block.ClusterBlocks;
-import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RoutingTable;
-import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.ShardsAllocation;
-import org.elasticsearch.common.collect.Sets;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -38,7 +35,6 @@ import org.elasticsearch.index.Index;
 import org.elasticsearch.indices.IndexMissingException;
 import org.elasticsearch.threadpool.ThreadPool;
 
-import java.util.Set;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -93,46 +89,26 @@ public class MetaDataDeleteIndexService extends AbstractComponent {
 
                     ClusterBlocks blocks = ClusterBlocks.builder().blocks(currentState.blocks()).removeIndexBlocks(request.index).build();
 
-                    // initialize the counter only for nodes the shards are allocated to
+                    final AtomicInteger counter = new AtomicInteger(currentState.nodes().size());
-                    Set<String> allocatedNodes = Sets.newHashSet();
+
-                    if (currentState.routingTable().hasIndex(request.index)) {
+                    final NodeIndexDeletedAction.Listener nodeIndexDeleteListener = new NodeIndexDeletedAction.Listener() {
-                        for (IndexShardRoutingTable indexShardRoutingTable : currentState.routingTable().index(request.index)) {
+                        @Override public void onNodeIndexDeleted(String index, String nodeId) {
-                            for (ShardRouting shardRouting : indexShardRoutingTable) {
+                            if (index.equals(request.index)) {
-                                if (shardRouting.currentNodeId() != null) {
+                                if (counter.decrementAndGet() == 0) {
-                                    allocatedNodes.add(shardRouting.currentNodeId());
+                                    listener.onResponse(new Response(true));
-                                }
+                                    nodeIndexDeletedAction.remove(this);
-                                if (shardRouting.relocatingNodeId() != null) {
-                                    allocatedNodes.add(shardRouting.relocatingNodeId());
                                 }
                             }
                         }
-                    }
+                    };
+                    nodeIndexDeletedAction.add(nodeIndexDeleteListener);
 
-                    if (allocatedNodes.isEmpty()) {
+                    listener.future = threadPool.schedule(request.timeout, ThreadPool.Names.SAME, new Runnable() {
-                        // no nodes allocated, don't wait for a response
+                        @Override public void run() {
-                        listener.onResponse(new Response(true));
+                            listener.onResponse(new Response(false));
-                    } else {
+                            nodeIndexDeletedAction.remove(nodeIndexDeleteListener);
-                        final AtomicInteger counter = new AtomicInteger(allocatedNodes.size());
+                        }
-
+                    });
-                        final NodeIndexDeletedAction.Listener nodeIndexDeleteListener = new NodeIndexDeletedAction.Listener() {
-                            @Override public void onNodeIndexDeleted(String index, String nodeId) {
-                                if (index.equals(request.index)) {
-                                    if (counter.decrementAndGet() == 0) {
-                                        listener.onResponse(new Response(true));
-                                        nodeIndexDeletedAction.remove(this);
-                                    }
-                                }
-                            }
-                        };
-                        nodeIndexDeletedAction.add(nodeIndexDeleteListener);
-
-                        listener.future = threadPool.schedule(request.timeout, ThreadPool.Names.SAME, new Runnable() {
-                            @Override public void run() {
-                                listener.onResponse(new Response(false));
-                                nodeIndexDeletedAction.remove(nodeIndexDeleteListener);
-                            }
-                        });
-                    }
 
                     return newClusterStateBuilder().state(currentState).routingResult(routingResult).metaData(newMetaData).blocks(blocks).build();
                 } catch (Exception e) {

modules/elasticsearch/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java

@@ -180,6 +180,24 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent<Indic
             applyDeletedShards(event);
             applyCleanedIndices(event);
             applySettings(event);
+            sendIndexLifecycleEvents(event);
+        }
+    }
+
+    private void sendIndexLifecycleEvents(final ClusterChangedEvent event) {
+        for (String index : event.indicesCreated()) {
+            try {
+                nodeIndexCreatedAction.nodeIndexCreated(index, event.state().nodes().localNodeId());
+            } catch (Exception e) {
+                logger.debug("failed to send to master index {} created event", index);
+            }
+        }
+        for (String index : event.indicesDeleted()) {
+            try {
+                nodeIndexDeletedAction.nodeIndexDeleted(index, event.state().nodes().localNodeId());
+            } catch (Exception e) {
+                logger.debug("failed to send to master index {} deleted event", index);
+            }
         }
     }
 
@@ -223,11 +241,6 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent<Indic
                 }
                 try {
                     indicesService.deleteIndex(index, "deleting index");
-                    threadPool.cached().execute(new Runnable() {
-                        @Override public void run() {
-                            nodeIndexDeletedAction.nodeIndexDeleted(index, event.state().nodes().localNodeId());
-                        }
-                    });
                 } catch (Exception e) {
                     logger.warn("failed to delete index", e);
                 }
@@ -292,11 +305,6 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent<Indic
                     logger.debug("[{}] creating index", indexMetaData.index());
                 }
                 indicesService.createIndex(indexMetaData.index(), indexMetaData.settings(), event.state().nodes().localNode().id());
-                threadPool.cached().execute(new Runnable() {
-                    @Override public void run() {
-                        nodeIndexCreatedAction.nodeIndexCreated(indexMetaData.index(), event.state().nodes().localNodeId());
-                    }
-                });
             }
         }
     }