Add Consistency Assertion to SnapshtosInProgress

Assert given input shards and indices are consistent.
Also, fixed the equality check for SnapshotsInProgress.
Before this change the tests never had more than a single waiting
shard per index so they never failed as a result of the
waiting shards list not being ordered.

Follow up to elastic#47552

Author: original-brownbear

server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

@@ -42,9 +42,12 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.snapshots.SnapshotInfo.METADATA_FIELD_INTRODUCED;
 
@@ -106,12 +109,25 @@ public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, Sta
             } else {
                 this.shards = shards;
                 this.waitingIndices = findWaitingIndices(shards);
+                assert assertShardsConsistent(state, indices, shards);
             }
             this.repositoryStateId = repositoryStateId;
             this.failure = failure;
             this.userMetadata = userMetadata;
         }
 
+        private static boolean assertShardsConsistent(State state, List<IndexId> indices,
+                                                      ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
+            if ((state == State.INIT || state == State.ABORTED) && shards.isEmpty()) {
+                return true;
+            }
+            final Set<String> indexNames = indices.stream().map(IndexId::getName).collect(Collectors.toSet());
+            final Set<String> indexNamesInShards = new HashSet<>();
+            shards.keysIt().forEachRemaining(s -> indexNamesInShards.add(s.getIndexName()));
+            assert indexNames.equals(indexNamesInShards)
+                : "Indices in shards " + indexNamesInShards + " differ from expected indices " + indexNames + " for state [" + state + "]";
+            return true;
+        }
         public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, State state, List<IndexId> indices,
                      long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards,
                      Map<String, Object> userMetadata) {
@@ -262,7 +278,10 @@ private ImmutableOpenMap<String, List<ShardId>> findWaitingIndices(ImmutableOpen
             Map<String, List<ShardId>> waitingIndicesMap = new HashMap<>();
             for (ObjectObjectCursor<ShardId, ShardSnapshotStatus> entry : shards) {
                 if (entry.value.state() == ShardState.WAITING) {
-                    waitingIndicesMap.computeIfAbsent(entry.key.getIndexName(), k -> new ArrayList<>()).add(entry.key);
+                    List<ShardId> waitingShards =
+                        waitingIndicesMap.computeIfAbsent(entry.key.getIndexName(), k -> new ArrayList<>());
+                    waitingShards.add(entry.key);
+                    waitingShards.sort(ShardId::compareTo);
                 }
             }
             if (waitingIndicesMap.isEmpty()) {

server/src/test/java/org/elasticsearch/snapshots/SnapshotsInProgressSerializationTests.java

@@ -36,6 +36,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;
 
 public class SnapshotsInProgressSerializationTests extends AbstractDiffableWireSerializationTestCase<Custom> {
 
@@ -62,13 +63,17 @@ private Entry randomSnapshot() {
         long startTime = randomLong();
         long repositoryStateId = randomLong();
         ImmutableOpenMap.Builder<ShardId, SnapshotsInProgress.ShardSnapshotStatus> builder = ImmutableOpenMap.builder();
-        int shardsCount = randomIntBetween(0, 10);
-        for (int j = 0; j < shardsCount; j++) {
-            ShardId shardId = new ShardId(new Index(randomAlphaOfLength(10), randomAlphaOfLength(10)), randomIntBetween(0, 10));
-            String nodeId = randomAlphaOfLength(10);
-            ShardState shardState = randomFrom(ShardState.values());
-            builder.put(shardId, new SnapshotsInProgress.ShardSnapshotStatus(nodeId, shardState,
-                shardState.failed() ? randomAlphaOfLength(10) : null, "1"));
+        final List<Index> esIndices =
+            indices.stream().map(i -> new Index(i.getName(), randomAlphaOfLength(10))).collect(Collectors.toList());
+        for (Index idx : esIndices) {
+            int shardsCount = randomIntBetween(1, 10);
+            for (int j = 0; j < shardsCount; j++) {
+                ShardId shardId = new ShardId(idx, j);
+                String nodeId = randomAlphaOfLength(10);
+                ShardState shardState = randomFrom(ShardState.values());
+                builder.put(shardId, new SnapshotsInProgress.ShardSnapshotStatus(nodeId, shardState,
+                    shardState.failed() ? randomAlphaOfLength(10) : null, "1"));
+            }
         }
         ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus> shards = builder.build();
         return new Entry(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards,