better state when closing shard, and handling its state

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/replication/ReplicationShardOperationFailedException.java

@@ -26,7 +26,7 @@ import org.elasticsearch.index.shard.ShardId;
 /**
  * An exception indicating that a failure occurred performing an operation on the shard.
  *
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public class ReplicationShardOperationFailedException extends IndexShardException implements ElasticSearchWrapperException {
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/replication/TransportShardReplicationOperationAction.java

@@ -42,7 +42,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.IndexShardMissingException;
 import org.elasticsearch.index.shard.IllegalIndexShardStateException;
-import org.elasticsearch.index.shard.IndexShardNotStartedException;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.shard.service.IndexShard;
 import org.elasticsearch.indices.IndexMissingException;
@@ -295,8 +294,9 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
                             }
 
                             @Override public void handleException(RemoteTransportException exp) {
-                                // if we got disconnected from the node, retry it...
+                                // if we got disconnected from the node, or the node / shard is not in the right state (being closed)
-                                if (exp.unwrapCause() instanceof ConnectTransportException || exp.unwrapCause() instanceof NodeCloseException) {
+                                if (exp.unwrapCause() instanceof ConnectTransportException || exp.unwrapCause() instanceof NodeCloseException ||
+                                        exp.unwrapCause() instanceof IllegalIndexShardStateException) {
                                     primaryOperationStarted.set(false);
                                     retryPrimary(fromClusterEvent, shard.shardId());
                                 } else {
@@ -379,8 +379,8 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
                 Response response = shardOperationOnPrimary(new ShardOperationRequest(primaryShardId, request));
                 performReplicas(response, alreadyThreaded);
             } catch (Exception e) {
-                if (e instanceof IndexShardMissingException || e instanceof IndexShardNotStartedException
+                // shard has not been allocated yet, retry it here
-                        || e instanceof IndexMissingException) {
+                if (e instanceof IndexShardMissingException || e instanceof IllegalIndexShardStateException || e instanceof IndexMissingException) {
                     retryPrimary(fromDiscoveryListener, shard.shardId());
                     return;
                 }
@@ -559,13 +559,7 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
         }
 
         /**
-         * Should an exception be ignored when the operation is performed on the replica. The exception
+         * Should an exception be ignored when the operation is performed on the replica.
-         * is ignored if it is:
-         *
-         * <ul>
-         * <li><tt>IllegalIndexShardStateException</tt>: The shard has not yet moved to started mode (it is still recovering).
-         * <li><tt>IndexMissingException</tt>/<tt>IndexShardMissingException</tt>: The shard has not yet started to initialize on the target node.
-         * </ul>
          */
         private boolean ignoreReplicaException(Throwable e) {
             Throwable cause = ExceptionsHelper.unwrapCause(e);

modules/elasticsearch/src/main/java/org/elasticsearch/index/engine/EngineClosedException.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.engine;
+
+import org.elasticsearch.index.shard.IndexShardClosedException;
+import org.elasticsearch.index.shard.ShardId;
+
+/**
+ * An engine is already closed.
+ *
+ * <p>Note, the relationship between shard and engine indicates that engine closed is shard closed, and
+ * we might get something slipping through the the shard and into the engine while the shard is closing.
+ *
+ * @author kimchy (shay.banon)
+ */
+public class EngineClosedException extends IndexShardClosedException {
+
+    public EngineClosedException(ShardId shardId) {
+        super(shardId);
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java

@@ -182,7 +182,11 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine,
     @Override public void create(Create create) throws EngineException {
         rwl.readLock().lock();
         try {
-            indexWriter.addDocument(create.doc(), create.analyzer());
+            IndexWriter writer = this.indexWriter;
+            if (writer == null) {
+                throw new EngineClosedException(shardId);
+            }
+            writer.addDocument(create.doc(), create.analyzer());
             translog.add(new Translog.Create(create));
             dirty = true;
         } catch (IOException e) {
@@ -195,7 +199,11 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine,
     @Override public void index(Index index) throws EngineException {
         rwl.readLock().lock();
         try {
-            indexWriter.updateDocument(index.uid(), index.doc(), index.analyzer());
+            IndexWriter writer = this.indexWriter;
+            if (writer == null) {
+                throw new EngineClosedException(shardId);
+            }
+            writer.updateDocument(index.uid(), index.doc(), index.analyzer());
             translog.add(new Translog.Index(index));
             dirty = true;
         } catch (IOException e) {
@@ -208,7 +216,11 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine,
     @Override public void delete(Delete delete) throws EngineException {
         rwl.readLock().lock();
         try {
-            indexWriter.deleteDocuments(delete.uid());
+            IndexWriter writer = this.indexWriter;
+            if (writer == null) {
+                throw new EngineClosedException(shardId);
+            }
+            writer.deleteDocuments(delete.uid());
             translog.add(new Translog.Delete(delete));
             dirty = true;
         } catch (IOException e) {
@@ -221,7 +233,11 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine,
     @Override public void delete(DeleteByQuery delete) throws EngineException {
         rwl.readLock().lock();
         try {
-            indexWriter.deleteDocuments(delete.query());
+            IndexWriter writer = this.indexWriter;
+            if (writer == null) {
+                throw new EngineClosedException(shardId);
+            }
+            writer.deleteDocuments(delete.query());
             translog.add(new Translog.DeleteByQuery(delete));
             dirty = true;
         } catch (IOException e) {