Avoid loading shard metadata while closing

server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java

@@ -41,7 +41,9 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.env.ShardLock;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.shard.ShardPath;
 import org.elasticsearch.index.shard.ShardStateMetaData;
@@ -52,6 +54,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 /**
  * This transport action is used to fetch the shard version from each node during primary allocation in {@link GatewayAllocator}.
@@ -116,55 +119,71 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) {
         try {
             final ShardId shardId = request.getShardId();
             logger.trace("{} loading local shard state info", shardId);
-            ShardStateMetaData shardStateMetaData = ShardStateMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY,
-                nodeEnv.availableShardPaths(request.shardId));
-            if (shardStateMetaData != null) {
-                IndexMetaData metaData = clusterService.state().metaData().index(shardId.getIndex());
-                if (metaData == null) {
-                    // we may send this requests while processing the cluster state that recovered the index
-                    // sometimes the request comes in before the local node processed that cluster state
-                    // in such cases we can load it from disk
-                    metaData = IndexMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY,
-                        nodeEnv.indexPaths(shardId.getIndex()));
-                }
-                if (metaData == null) {
-                    ElasticsearchException e = new ElasticsearchException("failed to find local IndexMetaData");
-                    e.setShard(request.shardId);
-                    throw e;
-                }
 
-                if (indicesService.getShardOrNull(shardId) == null) {
-                    // we don't have an open shard on the store, validate the files on disk are openable
-                    ShardPath shardPath = null;
-                    try {
-                        IndexSettings indexSettings = new IndexSettings(metaData, settings);
-                        shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings);
-                        if (shardPath == null) {
-                            throw new IllegalStateException(shardId + " no shard path found");
-                        }
-                        Store.tryOpenIndex(shardPath.resolveIndex(), shardId, nodeEnv::shardLock, logger);
-                    } catch (Exception exception) {
-                        final ShardPath finalShardPath = shardPath;
-                        logger.trace(() -> new ParameterizedMessage(
-                                "{} can't open index for shard [{}] in path [{}]",
-                                shardId,
-                                shardStateMetaData,
-                                (finalShardPath != null) ? finalShardPath.resolveIndex() : ""),
-                            exception);
-                        String allocationId = shardStateMetaData.allocationId != null ?
-                            shardStateMetaData.allocationId.getId() : null;
-                        return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary,
-                            exception);
-                    }
-                }
+            final IndexShard indexShard = indicesService.getShardOrNull(shardId);
+            if (indexShard != null) {
+                final ShardStateMetaData shardStateMetaData = indexShard.getShardStateMetaData();
+                final String allocationId = shardStateMetaData.allocationId != null ?
+                    shardStateMetaData.allocationId.getId() : null;
+                logger.trace("{} shard state info found: [{}]", shardId, shardStateMetaData);
+                return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary);
+            }
+
+            final ShardStateMetaData shardStateMetaData;
+            try (ShardLock ignored = nodeEnv.shardLock(shardId, TimeUnit.SECONDS.toMillis(5))) {
+                shardStateMetaData = ShardStateMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY,
+                    nodeEnv.availableShardPaths(shardId));
+            }
+
+            if (shardStateMetaData == null) {
+                logger.trace("{} no local shard info found", shardId);
+                return new NodeGatewayStartedShards(clusterService.localNode(), null, false);
+            }
 
-                logger.debug("{} shard state info found: [{}]", shardId, shardStateMetaData);
+            IndexMetaData metaData = clusterService.state().metaData().index(shardId.getIndex());
+            if (metaData == null) {
+                // we may send this requests while processing the cluster state that recovered the index
+                // sometimes the request comes in before the local node processed that cluster state
+                // in such cases we can load it from disk
+                metaData = IndexMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY,
+                    nodeEnv.indexPaths(shardId.getIndex()));
+            }
+            if (metaData == null) {
+                ElasticsearchException e = new ElasticsearchException("failed to find local IndexMetaData");
+                e.setShard(request.shardId);
+                throw e;
+            }
+
+            // we don't have an open shard on the store, validate the files on disk are openable
+            ShardPath shardPath = null;
+            try {
+                IndexSettings indexSettings = new IndexSettings(metaData, settings);
+                try (ShardLock ignored = nodeEnv.shardLock(shardId, TimeUnit.SECONDS.toMillis(5))) {
+                    shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings);
+                }
+                if (shardPath == null) {
+                    throw new IllegalStateException(shardId + " no shard path found");
+                }
+                Store.tryOpenIndex(shardPath.resolveIndex(), shardId, nodeEnv::shardLock, logger);
+            } catch (Exception exception) {
+                final ShardPath finalShardPath = shardPath;
+                logger.trace(() -> new ParameterizedMessage(
+                        "{} can't open index for shard [{}] in path [{}]",
+                        shardId,
+                        shardStateMetaData,
+                        (finalShardPath != null) ? finalShardPath.resolveIndex() : ""),
+                    exception);
                 String allocationId = shardStateMetaData.allocationId != null ?
                     shardStateMetaData.allocationId.getId() : null;
-                return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary);
+                return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary,
+                    exception);
             }
-            logger.trace("{} no local shard info found", shardId);
-            return new NodeGatewayStartedShards(clusterService.localNode(), null, false);
+
+            logger.debug("{} shard state info found: [{}]", shardId, shardStateMetaData);
+            String allocationId = shardStateMetaData.allocationId != null ?
+                shardStateMetaData.allocationId.getId() : null;
+            return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary);
+
         } catch (Exception e) {
             throw new ElasticsearchException("failed to load started shards", e);
         }

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -2065,6 +2065,12 @@ public void startRecovery(RecoveryState recoveryState, PeerRecoveryTargetService
         }
     }
 
+    public ShardStateMetaData getShardStateMetaData() {
+        synchronized (mutex) {
+            return new ShardStateMetaData(shardRouting.primary(), indexSettings.getUUID(), shardRouting.allocationId());
+        }
+    }
+
     /**
      * Returns whether the shard is in primary mode, i.e., in charge of replicating changes (see {@link ReplicationTracker}).
      */

server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java

@@ -41,6 +41,7 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.env.ShardLock;
 import org.elasticsearch.gateway.AsyncShardFetch;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.IndexSettings;
@@ -139,7 +140,10 @@ private StoreFilesMetaData listStoreMetaData(ShardId shardId) throws IOException
                 return new StoreFilesMetaData(shardId, Store.MetadataSnapshot.EMPTY);
             }
             final IndexSettings indexSettings = indexService != null ? indexService.getIndexSettings() : new IndexSettings(metaData, settings);
-            final ShardPath shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings);
+            final ShardPath shardPath;
+            try (ShardLock ignored = nodeEnv.shardLock(shardId, TimeUnit.SECONDS.toMillis(5))) {
+                shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings);
+            }
             if (shardPath == null) {
                 return new StoreFilesMetaData(shardId, Store.MetadataSnapshot.EMPTY);
             }

server/src/test/java/org/elasticsearch/gateway/RecoveryFromGatewayIT.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.query.QueryBuilders;
@@ -46,8 +47,10 @@
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.elasticsearch.test.InternalTestCluster;
 import org.elasticsearch.test.InternalTestCluster.RestartCallback;
+import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.test.store.MockFSIndexStore;
 
+import java.io.File;
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -58,6 +61,8 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.IntStream;
 
 import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS;
@@ -567,4 +572,70 @@ public Settings onNodeStopped(String nodeName) throws Exception {
         // start another node so cluster consistency checks won't time out due to the lack of state
         internalCluster().startNode();
     }
+
+    public void testLoadLatestStateWhileClosingShardDoesNotResurrectMetadataDirectory() throws Exception {
+
+        // This test pertains to a race condition in which deleting a shard concurrently with a TransportNodesListGatewayStartedShards
+        // request could resurrect the shard's metadata folder after it was deleted. Here we try and recreate the race, but it is quite
+        // delicate so this test does not always fail. Experimentation showed that setting the thread counts as below would yield a failure
+        // after a reasonable number of iterations: running repeatedly with -Dtests.iters=1000 saw 6 failures out of 10 runs.
+        //
+        // NB this experiment was run with
+        // @TestLogging("org.elasticsearch.env.NodeEnvironment:TRACE,org.elasticsearch.gateway.MetaDataStateFormat:TRACE," +
+        //     "org.elasticsearch.gateway.TransportNodesListGatewayStartedShards:TRACE,org.elasticsearch.index.shard.IndexShard:TRACE")
+        // but with less verbose logging the failures seem rarer.
+
+        final String nodeName = internalCluster().startNode();
+        DiscoveryNode node = internalCluster().getInstance(ClusterService.class, nodeName).localNode();
+
+        assertAcked(prepareCreate("test").setSettings(Settings.builder()
+            .put(SETTING_NUMBER_OF_SHARDS, 1).put(SETTING_NUMBER_OF_REPLICAS, 0)));
+        final ShardId shardId = new ShardId(resolveIndex("test"), 0);
+
+        final int listingThreadCount = 2;
+        final int deletingThreadCount = 2;
+
+        final CountDownLatch countDownLatch = new CountDownLatch(listingThreadCount + deletingThreadCount);
+
+        Thread threads[] = new Thread[listingThreadCount + deletingThreadCount];
+        for (int threadIndex = 0; threadIndex < listingThreadCount + deletingThreadCount; threadIndex++) {
+            final boolean isListingThread = threadIndex < listingThreadCount;
+            threads[threadIndex] = new Thread(() -> {
+                try {
+                    countDownLatch.countDown();
+                    countDownLatch.await();
+
+                    if (isListingThread) {
+                        internalCluster().getInstance(TransportNodesListGatewayStartedShards.class)
+                            .execute(new TransportNodesListGatewayStartedShards.Request(shardId, new DiscoveryNode[]{node}))
+                            .get();
+                    } else {
+                        assertAcked(client().admin().indices().prepareDelete("test"));
+                    }
+                } catch (InterruptedException | ExecutionException | IndexNotFoundException ignored) {
+                    // don't care if this fails
+                }
+            }, (isListingThread ? "Listing" : "Deleting") + "[" + threadIndex + "]");
+        }
+
+        NodeEnvironment nodeEnvironment = internalCluster().getInstance(NodeEnvironment.class, nodeName);
+
+        boolean directoryExists = false;
+        for (Path path : nodeEnvironment.availableShardPaths(shardId)) {
+            directoryExists = directoryExists || Files.exists(path);
+        }
+        assertTrue(directoryExists);
+
+        for (final Thread thread : threads) {
+            thread.start();
+        }
+
+        for (final Thread thread : threads) {
+            thread.join();
+        }
+
+        for (Path path : nodeEnvironment.availableShardPaths(shardId)) {
+            assertFalse(path + " should not exist", Files.exists(path));
+        }
+    }
 }