Revert "Remove multiple paths from ShardPath (#72713)" (#78485)

This reverts commit 08a89c89925eafb78fe61219af302d4b00c314b8.

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

@@ -160,7 +160,7 @@ public class RemoveCorruptedShardDataCommand extends ElasticsearchNodeCommand {
             .resolve(Integer.toString(shId.id()));
         if (Files.exists(shardPathLocation)) {
             final ShardPath shardPath = ShardPath.loadShardPath(logger, shId, indexSettings.customDataPath(),
-                shardPathLocation, dataPath);
+                new Path[]{shardPathLocation}, dataPath);
             if (shardPath != null) {
                 consumer.accept(shardPath);
             }

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

@@ -99,44 +99,59 @@ public final class ShardPath {
     }
 
     /**
-     * This method resolves the node's shard path using the given {@link NodeEnvironment}.
+     * This method walks through the nodes shard paths to find the data and state path for the given shard. If multiple
+     * directories with a valid shard state exist the one with the highest version will be used.
      * <b>Note:</b> this method resolves custom data locations for the shard if such a custom data path is provided.
      */
     public static ShardPath loadShardPath(Logger logger, NodeEnvironment env,
                                           ShardId shardId, String customDataPath) throws IOException {
         final Path shardPath = env.availableShardPath(shardId);
         final Path sharedDataPath = env.sharedDataPath();
-        return loadShardPath(logger, shardId, customDataPath, shardPath, sharedDataPath);
+        return loadShardPath(logger, shardId, customDataPath, new Path[] { shardPath }, sharedDataPath);
     }
 
     /**
-     * This method resolves the node's shard path using the given data paths.
+     * This method walks through the nodes shard paths to find the data and state path for the given shard. If multiple
+     * directories with a valid shard state exist the one with the highest version will be used.
      * <b>Note:</b> this method resolves custom data locations for the shard.
      */
-    public static ShardPath loadShardPath(Logger logger, ShardId shardId, String customDataPath, Path shardPath,
+    public static ShardPath loadShardPath(Logger logger, ShardId shardId, String customDataPath, Path[] availableShardPaths,
                                           Path sharedDataPath) throws IOException {
         final String indexUUID = shardId.getIndex().getUUID();
-        // EMPTY is safe here because we never call namedObject
-        ShardStateMetadata load = ShardStateMetadata.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY, shardPath);
-        if (load != null) {
-            if (load.indexUUID.equals(indexUUID) == false && IndexMetadata.INDEX_UUID_NA_VALUE.equals(load.indexUUID) == false) {
-                logger.warn("{} found shard on path: [{}] with a different index UUID - this "
-                    + "shard seems to be leftover from a different index with the same name. "
-                    + "Remove the leftover shard in order to reuse the path with the current index", shardId, shardPath);
-                throw new IllegalStateException(shardId + " index UUID in shard state was: " + load.indexUUID
-                    + " expected: " + indexUUID + " on shard path: " + shardPath);
+        Path loadedPath = null;
+        for (Path path : availableShardPaths) {
+            // EMPTY is safe here because we never call namedObject
+            ShardStateMetadata load = ShardStateMetadata.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY, path);
+            if (load != null) {
+                if (load.indexUUID.equals(indexUUID) == false && IndexMetadata.INDEX_UUID_NA_VALUE.equals(load.indexUUID) == false) {
+                    logger.warn("{} found shard on path: [{}] with a different index UUID - this "
+                        + "shard seems to be leftover from a different index with the same name. "
+                        + "Remove the leftover shard in order to reuse the path with the current index", shardId, path);
+                    throw new IllegalStateException(shardId + " index UUID in shard state was: " + load.indexUUID
+                        + " expected: " + indexUUID + " on shard path: " + path);
+                }
+                if (loadedPath == null) {
+                    loadedPath = path;
+                } else{
+                    throw new IllegalStateException(shardId + " more than one shard state found");
+                }
             }
+
+        }
+        if (loadedPath == null) {
+            return null;
+        } else {
             final Path dataPath;
+            final Path statePath = loadedPath;
             final boolean hasCustomDataPath = Strings.isNotEmpty(customDataPath);
             if (hasCustomDataPath) {
                 dataPath = NodeEnvironment.resolveCustomLocation(customDataPath, shardId, sharedDataPath);
             } else {
-                dataPath = shardPath;
+                dataPath = statePath;
             }
-            logger.debug("{} loaded data path [{}], state path [{}]", shardId, dataPath, shardPath);
-            return new ShardPath(hasCustomDataPath, dataPath, shardPath, shardId);
+            logger.debug("{} loaded data path [{}], state path [{}]", shardId, dataPath, statePath);
+            return new ShardPath(hasCustomDataPath, dataPath, statePath, shardId);
         }
-        return null;
     }
 
     /**