IndicesClusterStateService should replace an init. replica with an init. primary with the same aId

[bleskes]

In rare cases it is possible that a nodes gets an instruction to replace a replica
shard that's in POST_RECOVERY with a new initializing primary with the same allocation id.
This can happen by batching cluster states that include the starting of the replica, with
closing of the indices, opening it up again and allocating the primary shard to the node in
question. The node should then clean it's initializing replica and replace it with a new
initializing primary.

I'm not sure whether the test I added really adds enough value as existing tests found this. The main reason I added is to allow for simpler reproduction and to double check I fixed it. I'm open to discuss if we should keep.

Closes #32308

[elasticmachine]

Pinging @elastic/es-distributed

[ywelsch]

I've suggested an (imo simpler) alternative.

[ywelsch]

I think we're over-special-casing this, while also possibly missing some more murky cases. May I suggest the following alternative instead, which handles both this case as well as the previous (odd) one:

diff --git a/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java b/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
index e6a86d47f55..42a88383937 100644
--- a/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
+++ b/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
@@ -219,10 +219,10 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
 
         removeUnallocatedIndices(event); // also removes shards of removed indices
 
-        failMissingShards(state);
-
         removeShards(state);   // removes any local shards that doesn't match what the master expects
 
+        failMissingShards(state);
+
         updateIndices(event); // can also fail shards, but these are then guaranteed to be in failedShardsCache
 
         createIndices(state);
@@ -414,17 +414,12 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
                     logger.debug("{} removing shard (stale allocation id, stale {}, new {})", shardId,
                         currentRoutingEntry, newShardRouting);
                     indexService.removeShard(shardId.id(), "removing shard (stale copy)");
-                } else if (newShardRouting.initializing() && currentRoutingEntry.active()) {
+                } else if (newShardRouting.primary() &&
+                    state.metaData().index(shardId.getIndex()).primaryTerm(shardId.id()) != shard.getPrimaryTerm()) {
                     // this can happen if the node was isolated/gc-ed, rejoins the cluster and a new shard with the same allocation id
                     // is assigned to it. Batch cluster state processing or if shard fetching completes before the node gets a new cluster
                     // state may result in a new shard being initialized while having the same allocation id as the currently started shard.
-                    logger.debug("{} removing shard (not active, current {}, new {})", shardId, currentRoutingEntry, newShardRouting);
-                    indexService.removeShard(shardId.id(), "removing shard (stale copy)");
-                } else if (newShardRouting.primary() && currentRoutingEntry.primary() == false && newShardRouting.initializing()) {
-                    assert currentRoutingEntry.initializing() : currentRoutingEntry; // see above if clause
-                    // this can happen when cluster state batching batches activation of the shard, closing an index, reopening it
-                    // and assigning an initializing primary to this node
-                    logger.debug("{} removing shard (not active, current {}, new {})", shardId, currentRoutingEntry, newShardRouting);
+                    logger.debug("{} removing shard (stale primary term)", shardId);
                     indexService.removeShard(shardId.id(), "removing shard (stale copy)");
                 }
             }
@@ -728,6 +723,11 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
          */
         ShardRouting routingEntry();
 
+        /**
+         * Returns the current primary term of this shard
+         */
+        long getPrimaryTerm();
+
         /**
          * Returns the latest internal shard state.
          */
diff --git a/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java b/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java
index 5e9ba653d40..bfcca1946a6 100644
--- a/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java
+++ b/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java
@@ -383,6 +383,11 @@ public abstract class AbstractIndicesClusterStateServiceTestCase extends ESTestC
             return shardRouting;
         }
 
+        @Override
+        public long getPrimaryTerm() {
+            return term;
+        }
+
         @Override
         public IndexShardState state() {
             return null;

[ywelsch]

After discussion, we agreed this is not quite right. We decided to go with the initial condition proposed here.

[ywelsch]

My previous suggestion was too good to be true. The condition you've proposed is the best I can think of right now until we distinguish between allocation ids and shard copy ids. I also think it's ok to keep the test, it's relatively short and succinctly documents an edge case.

[ywelsch]

After discussion, we agreed this is not quite right. We decided to go with the initial condition proposed here.

[ywelsch]

spelling

[bleskes]

Thx @ywelsch . I tend towards pushing this to 6.3 too. Do you agree?

[ywelsch]

yes

[bleskes]

thx @ywelsch