add isRelocated flag to simplify code reasoning

Author: ywelsch

server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -983,7 +983,7 @@ class PrimaryShardReference extends ShardReference
         }
 
         public boolean isRelocated() {
-            return indexShard.isPrimaryMode() == false;
+            return indexShard.isRelocatedPrimary();
         }
 
         @Override

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

@@ -85,6 +85,7 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
      *   computation from that point on.
      */
     volatile boolean primaryMode;
+
     /**
      * Boolean flag that indicates if a relocation handoff is in progress. A handoff is started by calling {@link #startRelocationHandoff}
      * and is finished by either calling {@link #completeRelocationHandoff} or {@link #abortRelocationHandoff}, depending on whether the
@@ -102,6 +103,11 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
      */
     boolean handoffInProgress;
 
+    /**
+     * Boolean flag that indicates whether a relocation handoff completed (see {@link #completeRelocationHandoff}).
+     */
+    volatile boolean relocated;
+
     /**
      * The global checkpoint tracker relies on the property that cluster state updates are applied in-order. After transferring a primary
      * context from the primary relocation source to the target and initializing the target, it is possible for the target to apply a
@@ -260,6 +266,13 @@ public boolean isPrimaryMode() {
         return primaryMode;
     }
 
+    /**
+     * Returns whether the replication tracker has relocated away to another shard copy.
+     */
+    public boolean isRelocated() {
+        return relocated;
+    }
+
     /**
      * Class invariant that should hold before and after every invocation of public methods on this class. As Java lacks implication
      * as a logical operator, many of the invariants are written under the form (!A || B), they should be read as (A implies B) however.
@@ -287,6 +300,9 @@ private boolean invariant() {
         // relocation handoff can only occur in primary mode
         assert !handoffInProgress || primaryMode;
 
+        // a relocated copy is not in primary mode
+        assert !relocated || !primaryMode;
+
         // the current shard is marked as in-sync when the global checkpoint tracker operates in primary mode
         assert !primaryMode || checkpoints.get(shardAllocationId).inSync;
 
@@ -766,8 +782,10 @@ public synchronized void completeRelocationHandoff() {
         assert invariant();
         assert primaryMode;
         assert handoffInProgress;
+        assert relocated == false;
         primaryMode = false;
         handoffInProgress = false;
+        relocated = true;
         // forget all checkpoint information except for global checkpoint of current shard
         checkpoints.entrySet().stream().forEach(e -> {
             final CheckpointState cps = e.getValue();

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

@@ -425,18 +425,11 @@ public void updateShardState(final ShardRouting newRouting,
                     "a primary relocation is completed by the master, but primary mode is not active " + currentRouting;
 
                 changeState(IndexShardState.STARTED, "global state is [" + newRouting.state() + "]");
-            } else if (currentRouting.primary() && currentRouting.relocating() &&
-                operationPrimaryTerm == pendingPrimaryTerm &&
-                replicationTracker.isPrimaryMode() == false &&
+            } else if (currentRouting.primary() && currentRouting.relocating() && replicationTracker.isRelocated() &&
                 (newRouting.relocating() == false || newRouting.equalsIgnoringMetaData(currentRouting) == false)) {
                 // if the shard is not in primary mode anymore (after primary relocation) we have to fail when any changes in shard routing occur (e.g. due to recovery
                 // failure / cancellation). The reason is that at the moment we cannot safely reactivate primary mode without risking two
                 // active primaries.
-                // We check for operationPrimaryTerm to be equal to pendingPrimaryTerm, which ensures that we have a fully baked primary,
-                // i.e. a primary that has has transitioned to primary mode. Assume for example an active replica, which then got a
-                // cluster state where it became promoted to relocating primary. This means that we asynchronously start the transition to
-                // primary in the background. A follow-up cluster state might then cancel relocation before we have completed the transition
-                // to primary, which would result in the shard to be failed if we did not check for the operationPrimaryTerm.
                 throw new IndexShardRelocatedException(shardId(), "Shard is marked as relocated, cannot safely move to state " + newRouting.state());
             }
             assert newRouting.active() == false || state == IndexShardState.STARTED || state == IndexShardState.CLOSED :
@@ -610,7 +603,7 @@ public void relocated(final Consumer<ReplicationTracker.PrimaryContext> consumer
                     consumer.accept(primaryContext);
                     synchronized (mutex) {
                         verifyRelocatingState();
-                        replicationTracker.completeRelocationHandoff(); // make changes to primaryMode flag only under mutex
+                        replicationTracker.completeRelocationHandoff(); // make changes to primaryMode and relocated flag only under mutex
                     }
                 } catch (final Exception e) {
                     try {
@@ -2113,10 +2106,11 @@ public void startRecovery(RecoveryState recoveryState, PeerRecoveryTargetService
     }
 
     /**
-     * Returns whether the shard is in primary mode, i.e., in charge of replicating changes (see {@link ReplicationTracker}).
+     * Returns whether the shard is a relocated primary, i.e. not in charge anymore of replicating changes (see {@link ReplicationTracker}).
      */
-    public boolean isPrimaryMode() {
-        return replicationTracker.isPrimaryMode();
+    public boolean isRelocatedPrimary() {
+        assert shardRouting.primary() : "only call isRelocatedPrimary on primary shard";
+        return replicationTracker.isRelocated();
     }
 
     class ShardEventListener implements Engine.EventListener {

server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java

@@ -250,7 +250,7 @@ public void onFailure(Exception e) {
             try (Releasable ignored = FutureUtils.get(permit)) {
                 // check that the IndexShard still has the primary authority. This needs to be checked under operation permit to prevent
                 // races, as IndexShard will switch its authority only when it holds all operation permits, see IndexShard.relocated()
-                if (primary.isPrimaryMode() == false) {
+                if (primary.isRelocatedPrimary()) {
                     throw new IndexShardRelocatedException(primary.shardId());
                 }
                 runnable.run();

server/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationActionTests.java

@@ -683,7 +683,7 @@ public void testSeqNoIsSetOnPrimary() throws Exception {
         final IndexShard shard = mock(IndexShard.class);
         when(shard.getPendingPrimaryTerm()).thenReturn(primaryTerm);
         when(shard.routingEntry()).thenReturn(routingEntry);
-        when(shard.isPrimaryMode()).thenReturn(true);
+        when(shard.isRelocatedPrimary()).thenReturn(false);
         IndexShardRoutingTable shardRoutingTable = clusterService.state().routingTable().shardRoutingTable(shardId);
         Set<String> inSyncIds = randomBoolean() ? Collections.singleton(routingEntry.allocationId().getId()) :
             clusterService.state().metaData().index(index).inSyncAllocationIds(0);
@@ -1217,7 +1217,7 @@ private IndexShard mockIndexShard(ShardId shardId, ClusterService clusterService
             }
             return routing;
         });
-        when(indexShard.isPrimaryMode()).thenAnswer(invocationOnMock -> isRelocated.get() == false);
+        when(indexShard.isRelocatedPrimary()).thenAnswer(invocationOnMock -> isRelocated.get());
         doThrow(new AssertionError("failed shard is not supported")).when(indexShard).failShard(anyString(), any(Exception.class));
         when(indexShard.getPendingPrimaryTerm()).thenAnswer(i ->
             clusterService.state().metaData().getIndexSafe(shardId.getIndex()).primaryTerm(shardId.id()));

server/src/test/java/org/elasticsearch/action/support/replication/TransportWriteActionTests.java

@@ -472,7 +472,7 @@ private IndexShard mockIndexShard(ShardId shardId, ClusterService clusterService
             }
             return routing;
         });
-        when(indexShard.isPrimaryMode()).thenAnswer(invocationOnMock -> isRelocated.get() == false);
+        when(indexShard.isRelocatedPrimary()).thenAnswer(invocationOnMock -> isRelocated.get());
         doThrow(new AssertionError("failed shard is not supported")).when(indexShard).failShard(anyString(), any(Exception.class));
         when(indexShard.getPendingPrimaryTerm()).thenAnswer(i ->
             clusterService.state().metaData().getIndexSafe(shardId.getIndex()).primaryTerm(shardId.id()));

server/src/test/java/org/elasticsearch/index/seqno/ReplicationTrackerTests.java

@@ -770,8 +770,10 @@ public void testPrimaryContextHandoff() throws IOException {
         assertThat(newPrimary.routingTable, equalTo(oldPrimary.routingTable));
         assertThat(newPrimary.replicationGroup, equalTo(oldPrimary.replicationGroup));
 
+        assertFalse(oldPrimary.relocated);
         oldPrimary.completeRelocationHandoff();
         assertFalse(oldPrimary.primaryMode);
+        assertTrue(oldPrimary.relocated);
     }
 
     public void testIllegalStateExceptionIfUnknownAllocationId() {

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -1421,14 +1421,14 @@ public void testLockingBeforeAndAfterRelocated() throws Exception {
             recoveryThread.start();
             latch.await();
             // recovery can only be finalized after we release the current primaryOperationLock
-            assertTrue(shard.isPrimaryMode());
+            assertFalse(shard.isRelocatedPrimary());
         }
         // recovery can be now finalized
         recoveryThread.join();
-        assertFalse(shard.isPrimaryMode());
+        assertTrue(shard.isRelocatedPrimary());
         try (Releasable ignored = acquirePrimaryOperationPermitBlockingly(shard)) {
             // lock can again be acquired
-            assertFalse(shard.isPrimaryMode());
+            assertTrue(shard.isRelocatedPrimary());
         }
 
         closeShards(shard);
@@ -1470,7 +1470,7 @@ public void onResponse(Releasable releasable) {
 
     public void testStressRelocated() throws Exception {
         final IndexShard shard = newStartedShard(true);
-        assertTrue(shard.isPrimaryMode());
+        assertFalse(shard.isRelocatedPrimary());
         IndexShardTestCase.updateRoutingEntry(shard, ShardRoutingHelper.relocate(shard.routingEntry(), "other_node"));
         final int numThreads = randomIntBetween(2, 4);
         Thread[] indexThreads = new Thread[numThreads];
@@ -1506,14 +1506,14 @@ public void run() {
         assertThat(relocated.get(), equalTo(false));
         assertThat(shard.getActiveOperationsCount(), greaterThan(0));
         // ensure we only transition after pending operations completed
-        assertTrue(shard.isPrimaryMode());
+        assertFalse(shard.isRelocatedPrimary());
         // complete pending operations
         barrier.await();
         // complete recovery/relocation
         recoveryThread.join();
         // ensure relocated successfully once pending operations are done
         assertThat(relocated.get(), equalTo(true));
-        assertFalse(shard.isPrimaryMode());
+        assertTrue(shard.isRelocatedPrimary());
         assertThat(shard.getActiveOperationsCount(), equalTo(0));
 
         for (Thread indexThread : indexThreads) {
@@ -1577,7 +1577,7 @@ protected void doRun() throws Exception {
         cyclicBarrier.await();
         relocationThread.join();
         cancellingThread.join();
-        if (shard.isPrimaryMode() == false) {
+        if (shard.isRelocatedPrimary()) {
             logger.debug("shard was relocated successfully");
             assertThat(cancellingException.get(), instanceOf(IllegalIndexShardStateException.class));
             assertThat("current routing:" + shard.routingEntry(), shard.routingEntry().relocating(), equalTo(true));
@@ -1858,7 +1858,7 @@ public void testRecoveryFailsAfterMovingToRelocatedState() throws InterruptedExc
         ShardRouting inRecoveryRouting = ShardRoutingHelper.relocate(origRouting, "some_node");
         IndexShardTestCase.updateRoutingEntry(shard, inRecoveryRouting);
         shard.relocated(primaryContext -> {});
-        assertFalse(shard.isPrimaryMode());
+        assertTrue(shard.isRelocatedPrimary());
         try {
             IndexShardTestCase.updateRoutingEntry(shard, origRouting);
             fail("Expected IndexShardRelocatedException");

server/src/test/java/org/elasticsearch/indices/recovery/RecoverySourceHandlerTests.java

@@ -395,7 +395,7 @@ public void testThrowExceptionOnPrimaryRelocatedBeforePhase1Started() throws IOE
         final IndexShard shard = mock(IndexShard.class);
         when(shard.seqNoStats()).thenReturn(mock(SeqNoStats.class));
         when(shard.segmentStats(anyBoolean())).thenReturn(mock(SegmentsStats.class));
-        when(shard.isPrimaryMode()).thenReturn(false);
+        when(shard.isRelocatedPrimary()).thenReturn(true);
         when(shard.acquireSafeIndexCommit()).thenReturn(mock(Engine.IndexCommitRef.class));
         doAnswer(invocation -> {
             ((ActionListener<Releasable>)invocation.getArguments()[0]).onResponse(() -> {});
@@ -444,7 +444,7 @@ public void testCancellationsDoesNotLeakPrimaryPermits() throws Exception {
         final CancellableThreads cancellableThreads = new CancellableThreads();
         final IndexShard shard = mock(IndexShard.class);
         final AtomicBoolean freed = new AtomicBoolean(true);
-        when(shard.isPrimaryMode()).thenReturn(true);
+        when(shard.isRelocatedPrimary()).thenReturn(false);
         doAnswer(invocation -> {
             freed.set(false);
             ((ActionListener<Releasable>)invocation.getArguments()[0]).onResponse(() -> freed.set(true));