improve logging when trying to delete unallocated shard, check first if deleting is required, and only then delete it

modules/elasticsearch/src/main/java/org/elasticsearch/index/store/IndexStore.java

@@ -52,5 +52,14 @@ public interface IndexStore extends IndexComponent {
      */
     ByteSizeValue backingStoreFreeSpace();
 
+    /**
+     * Returns <tt>true</tt> if this shard is allocated on this node. Allocated means
+     * that it has storage files that can be deleted using {@link #deleteUnallocated(org.elasticsearch.index.shard.ShardId)}.
+     */
+    boolean canDeleteUnallocated(ShardId shardId);
+
+    /**
+     * Deletes this shard store since its no longer allocated.
+     */
     void deleteUnallocated(ShardId shardId) throws IOException;
 }

modules/elasticsearch/src/main/java/org/elasticsearch/index/store/fs/FsIndexStore.java

@@ -78,6 +78,16 @@ public abstract class FsIndexStore extends AbstractIndexStore {
         return new ByteSizeValue(usableSpace);
     }
 
+    @Override public boolean canDeleteUnallocated(ShardId shardId) {
+        if (location == null) {
+            return false;
+        }
+        if (indexService.hasShard(shardId.id())) {
+            return false;
+        }
+        return shardLocation(shardId).exists();
+    }
+
     @Override public void deleteUnallocated(ShardId shardId) throws IOException {
         if (location == null) {
             return;

modules/elasticsearch/src/main/java/org/elasticsearch/index/store/support/AbstractIndexStore.java

@@ -41,6 +41,10 @@ public abstract class AbstractIndexStore extends AbstractIndexComponent implemen
         this.indexService = indexService;
     }
 
+    @Override public boolean canDeleteUnallocated(ShardId shardId) {
+        return false;
+    }
+
     @Override public void deleteUnallocated(ShardId shardId) throws IOException {
         // do nothing here...
     }

modules/elasticsearch/src/main/java/org/elasticsearch/indices/store/IndicesStore.java

@@ -74,6 +74,9 @@ public class IndicesStore extends AbstractComponent implements ClusterStateListe
                 if (indexService.hasShard(indexShardRoutingTable.shardId().id())) {
                     continue;
                 }
+                if (!indexService.store().canDeleteUnallocated(indexShardRoutingTable.shardId())) {
+                    continue;
+                }
                 // only delete an unallocated shard if all (other shards) are started
                 if (indexShardRoutingTable.countWithState(ShardRoutingState.STARTED) == indexShardRoutingTable.size()) {
                     if (logger.isDebugEnabled()) {