Ensure translog upload completes before primary mode change & reset new primary engine

Signed-off-by: Ashish Singh <ssashish@amazon.com>

Author: ashking94

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

@@ -762,12 +762,20 @@ public void relocated(
         final String targetAllocationId,
         final Consumer<ReplicationTracker.PrimaryContext> consumer,
         final Consumer<StepListener> performSegRep,
-        final ActionListener<Void> listener
+        final ActionListener<Void> listener,
+        final boolean syncTranslog
     ) throws IllegalIndexShardStateException, IllegalStateException, InterruptedException {
         assert shardRouting.primary() : "only primaries can be marked as relocated: " + shardRouting;
         try (Releasable forceRefreshes = refreshListeners.forceRefreshes()) {
             indexShardOperationPermits.blockOperations(30, TimeUnit.MINUTES, () -> {
                 forceRefreshes.close();
+
+                // Since all the index permits are acquired at this point, the translog buffer will not change.
+                // It is safe to perform sync of translogs now as this will ensure for remote-backed indexes, the
+                // translogs has been uploaded to the remote store.
+                if (syncTranslog) {
+                    maybeSync();
+                }
                 // no shard operation permits are being held here, move state from started to relocated
                 assert indexShardOperationPermits.getActiveOperationsCount() == OPERATIONS_BLOCKED
                     : "in-flight operations in progress while moving shard state to relocated";
@@ -808,6 +816,16 @@ public void relocated(
         }
     }
 
+    private void maybeSync() {
+        try {
+            if (isSyncNeeded()) {
+                sync();
+            }
+        } catch (IOException e) {
+            logger.warn("failed to sync translog", e);
+        }
+    }
+
     private void verifyRelocatingState() {
         if (state != IndexShardState.STARTED) {
             throw new IndexShardNotStartedException(shardId, state);

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

@@ -58,6 +58,7 @@
 import org.opensearch.common.util.concurrent.ListenableFuture;
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.core.internal.io.IOUtils;
+import org.opensearch.index.IndexSettings;
 import org.opensearch.index.engine.RecoveryEngineException;
 import org.opensearch.index.seqno.RetentionLease;
 import org.opensearch.index.seqno.RetentionLeaseNotFoundException;
@@ -821,17 +822,21 @@ void finalizeRecovery(long targetLocalCheckpoint, long trimAboveSeqNo, ActionLis
             final StepListener<Void> handoffListener = new StepListener<>();
             if (request.isPrimaryRelocation()) {
                 logger.trace("performing relocation hand-off");
-                final Consumer<StepListener> forceSegRepConsumer = shard.indexSettings().isSegRepEnabled()
+                final IndexSettings indexSettings = shard.indexSettings();
+                final Consumer<StepListener> forceSegRepConsumer = indexSettings.isSegRepEnabled()
                     ? recoveryTarget::forceSegmentFileSync
                     : res -> res.onResponse(null);
+                final boolean syncTranslog = indexSettings.isRemoteTranslogStoreEnabled()
+                    && Translog.Durability.ASYNC == indexSettings.getTranslogDurability();
                 // TODO: make relocated async
                 // this acquires all IndexShard operation permits and will thus delay new recoveries until it is done
                 cancellableThreads.execute(
                     () -> shard.relocated(
                         request.targetAllocationId(),
                         recoveryTarget::handoffPrimaryContext,
                         forceSegRepConsumer,
-                        handoffListener
+                        handoffListener,
+                        syncTranslog
                     )
                 );
                 /*

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

@@ -1011,7 +1011,8 @@ public void onResponse(Void unused) {
                         public void onFailure(Exception e) {
                             fail(e.toString());
                         }
-                    }
+                    },
+                    false
                 );
                 engineClosed = false;
                 break;
@@ -2039,7 +2040,8 @@ public void onResponse(Void unused) {
                         public void onFailure(Exception e) {
                             fail(e.toString());
                         }
-                    }
+                    },
+                    false
                 );
             } catch (InterruptedException e) {
                 throw new RuntimeException(e);
@@ -2086,7 +2088,8 @@ public void onResponse(Void unused) {
                         public void onFailure(Exception e) {
                             fail(e.toString());
                         }
-                    }
+                    },
+                    false
                 );
             } catch (InterruptedException e) {
                 throw new RuntimeException(e);
@@ -2190,7 +2193,8 @@ public void onFailure(Exception e) {
                             relocated.set(false);
                             fail(e.toString());
                         }
-                    }
+                    },
+                    false
                 );
             } catch (InterruptedException e) {
                 throw new RuntimeException(e);
@@ -2239,7 +2243,8 @@ public void onResponse(Void unused) {
                 public void onFailure(Exception e) {
                     fail(e.toString());
                 }
-            }
+            },
+            false
         );
         expectThrows(IllegalIndexShardStateException.class, () -> IndexShardTestCase.updateRoutingEntry(shard, originalRouting));
         closeShards(shard);
@@ -2265,7 +2270,8 @@ public void onFailure(Exception e) {
                     assertTrue(ExceptionsHelper.unwrapCause(e) instanceof ReplicationFailedException);
                     assertEquals(e.getMessage(), "Segment replication failed");
                 }
-            }
+            },
+            false
         );
         closeShards(shard);
     }
@@ -2290,7 +2296,8 @@ public void onResponse(Void unused) {
                 public void onFailure(Exception e) {
                     assertTrue(ExceptionsHelper.unwrapCause(e) instanceof IllegalIndexShardStateException);
                 }
-            }
+            },
+            false
         );
         closeShards(shard);
     }
@@ -2325,7 +2332,8 @@ public void onResponse(Void unused) {
                         public void onFailure(Exception e) {
                             relocationException.set(e);
                         }
-                    }
+                    },
+                    false
                 );
             }
         });
@@ -2396,7 +2404,8 @@ public void onFailure(Exception e) {
                         )
                     );
                 }
-            }
+            },
+            false
         );
 
         assertFalse(relocated.get());
@@ -2414,7 +2423,8 @@ public void onResponse(Void unused) {
                 public void onFailure(Exception e) {
                     fail(e.toString());
                 }
-            }
+            },
+            false
         );
         assertTrue(relocated.get());
         closeShards(shard);
@@ -2822,7 +2832,8 @@ public void onResponse(Void unused) {
                 public void onFailure(Exception e) {
                     fail(e.toString());
                 }
-            }
+            },
+            false
         );
         assertTrue(shard.isRelocatedPrimary());
         try {