Only run retention lease actions on active primary (#40386)

In some cases, a request to perform a retention lease action can arrive
on a primary shard before it is active. In this case, the primary shard
would not yet be in primary mode, tripping an assertion in the
replication tracker. Instead, we should not attempt to perform such
actions on an initializing shard. This commit addresses this by not
returning the primary shard in the single shard iterator if the primary
shard is not yet active.

Author: jasontedor

server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseActions.java

@@ -28,6 +28,8 @@
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
+import org.elasticsearch.cluster.routing.PlainShardIterator;
 import org.elasticsearch.cluster.routing.ShardsIterator;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
@@ -42,6 +44,7 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.Objects;
 import java.util.function.Supplier;
 
@@ -84,10 +87,14 @@ abstract static class TransportRetentionLeaseAction<T extends Request<T>> extend
 
         @Override
         protected ShardsIterator shards(final ClusterState state, final InternalRequest request) {
-            return state
+            final IndexShardRoutingTable shardRoutingTable = state
                     .routingTable()
-                    .shardRoutingTable(request.concreteIndex(), request.request().getShardId().id())
-                    .primaryShardIt();
+                    .shardRoutingTable(request.concreteIndex(), request.request().getShardId().id());
+            if (shardRoutingTable.primaryShard().active()) {
+                return shardRoutingTable.primaryShardIt();
+            } else {
+                return new PlainShardIterator(request.request().getShardId(), Collections.emptyList());
+            }
         }
 
         @Override

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRetentionLeaseIT.java

@@ -43,7 +43,6 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.snapshots.RestoreInfo;
 import org.elasticsearch.snapshots.RestoreService;
-import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.RemoteTransportException;
@@ -266,7 +265,6 @@ public void testRetentionLeaseIsRenewedDuringRecovery() throws Exception {
 
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40089")
     public void testRetentionLeasesAreNotBeingRenewedAfterRecoveryCompletes() throws Exception {
         final String leaderIndex = "leader";
         final int numberOfShards = randomIntBetween(1, 3);
@@ -463,7 +461,6 @@ public void testUnfollowRemovesRetentionLeases() throws Exception {
         }
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40089")
     public void testUnfollowFailsToRemoveRetentionLeases() throws Exception {
         final String leaderIndex = "leader";
         final String followerIndex = "follower";
@@ -534,7 +531,6 @@ public void testUnfollowFailsToRemoveRetentionLeases() throws Exception {
         }
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40089")
     public void testRetentionLeaseRenewedWhileFollowing() throws Exception {
         final String leaderIndex = "leader";
         final String followerIndex = "follower";
@@ -618,7 +614,6 @@ public void testRetentionLeaseAdvancesWhileFollowing() throws Exception {
     }
 
     @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/39509")
-    @TestLogging(value = "org.elasticsearch.xpack.ccr:trace")
     public void testRetentionLeaseRenewalIsCancelledWhenFollowingIsPaused() throws Exception {
         final String leaderIndex = "leader";
         final String followerIndex = "follower";
@@ -748,7 +743,6 @@ public void testRetentionLeaseRenewalIsResumedWhenFollowingIsResumed() throws Ex
         assertRetentionLeaseRenewal(numberOfShards, numberOfReplicas, followerIndex, leaderIndex);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40089")
     public void testRetentionLeaseIsAddedIfItDisappearsWhileFollowing() throws Exception {
         final String leaderIndex = "leader";
         final String followerIndex = "follower";