smarter handling of throttling when preferring allocating to an existing work location

modules/elasticsearch/src/main/java/org/elasticsearch/gateway/blobstore/BlobReuseExistingNodeAllocation.java

@@ -118,7 +118,9 @@ public class BlobReuseExistingNodeAllocation extends NodeAllocation {
                 }
 
                 // check if we can allocate on that node...
-                if (!nodeAllocations.canAllocate(shard, node, routingNodes).allocate()) {
+                // we only check for NO, since if this node is THROTTLING and it has enough "same data"
+                // then we will try and assign it next time
+                if (nodeAllocations.canAllocate(shard, node, routingNodes) == Decision.NO) {
                     continue;
                 }
 

modules/elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexShardGatewayService.java

@@ -192,6 +192,11 @@ public class IndexShardGatewayService extends AbstractIndexShardComponent implem
                 } catch (IndexShardNotStartedException e) {
                     listener.onIgnoreRecovery("shard closed");
                 } catch (Exception e) {
+                    if (indexShard.state() == IndexShardState.CLOSED) {
+                        // got closed on us, just ignore this recovery
+                        listener.onIgnoreRecovery("shard closed");
+                        return;
+                    }
                     listener.onRecoveryFailed(new IndexShardGatewayRecoveryException(shardId, "failed recovery", e));
                 }
             }

modules/test/integration/src/test/java/org/elasticsearch/test/integration/gateway/AbstractSimpleIndexGatewayTests.java

@@ -220,6 +220,10 @@ public abstract class AbstractSimpleIndexGatewayTests extends AbstractNodesTests
         testLoad(false);
     }
 
+    protected boolean isPersistentStorage() {
+        return true;
+    }
+
     private void testLoad(boolean fullRecovery) {
         startNode("server1");
 
@@ -265,7 +269,7 @@ public abstract class AbstractSimpleIndexGatewayTests extends AbstractNodesTests
         for (IndexShardStatus indexShardStatus : statusResponse.index("test")) {
             for (ShardStatus shardStatus : indexShardStatus) {
                 if (shardStatus.shardRouting().primary()) {
-                    if (fullRecovery) {
+                    if (fullRecovery || !isPersistentStorage()) {
                         assertThat(shardStatus.gatewayRecoveryStatus().reusedIndexSize().bytes(), equalTo(0l));
                     } else {
                         assertThat(shardStatus.gatewayRecoveryStatus().reusedIndexSize().bytes(), greaterThan(shardStatus.gatewayRecoveryStatus().indexSize().bytes() - 4098 /* segments file */));

modules/test/integration/src/test/java/org/elasticsearch/test/integration/gateway/fs/SimpleFsIndexInRamIndexGatewayTests.java

@@ -26,4 +26,7 @@ import org.elasticsearch.test.integration.gateway.AbstractSimpleIndexGatewayTest
  */
 public class SimpleFsIndexInRamIndexGatewayTests extends AbstractSimpleIndexGatewayTests {
 
+    @Override protected boolean isPersistentStorage() {
+        return false;
+    }
 }