From 95afefdbb911a1c3565e5fc2c2cea1f6d2809755 Mon Sep 17 00:00:00 2001
From: Armin Braun <me@obrown.io>
Date: Mon, 7 Oct 2019 08:38:20 +0200
Subject: [PATCH] Add Consistency Assertion to SnapshotsInProgress (#47598)

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 #47552
---
 .../cluster/SnapshotsInProgress.java          | 18 ++++++++++++++++--
 ...SnapshotsInProgressSerializationTests.java | 19 ++++++++++++-------
 2 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java b/server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java
index 6ae6a8109ddd0..d30f488c5d70d 100644
--- a/server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java
+++ b/server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java
@@ -41,9 +41,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;
 
@@ -105,12 +108,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) {
@@ -189,7 +205,6 @@ public boolean equals(Object o) {
             if (!shards.equals(entry.shards)) return false;
             if (!snapshot.equals(entry.snapshot)) return false;
             if (state != entry.state) return false;
-            if (!waitingIndices.equals(entry.waitingIndices)) return false;
             if (repositoryStateId != entry.repositoryStateId) return false;
 
             return true;
@@ -203,7 +218,6 @@ public int hashCode() {
             result = 31 * result + (partial ? 1 : 0);
             result = 31 * result + shards.hashCode();
             result = 31 * result + indices.hashCode();
-            result = 31 * result + waitingIndices.hashCode();
             result = 31 * result + Long.hashCode(startTime);
             result = 31 * result + Long.hashCode(repositoryStateId);
             return result;
diff --git a/server/src/test/java/org/elasticsearch/snapshots/SnapshotsInProgressSerializationTests.java b/server/src/test/java/org/elasticsearch/snapshots/SnapshotsInProgressSerializationTests.java
index 46404af9f4037..03a1e96ea2a89 100644
--- a/server/src/test/java/org/elasticsearch/snapshots/SnapshotsInProgressSerializationTests.java
+++ b/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));
+        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));
+            }
         }
         ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus> shards = builder.build();
         return new Entry(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards,