[Remote Store] Add Primary/Replica side changes to support Dual Repli…

…cation during Remote Store Migration (#12821)

Signed-off-by: Shourya Dutta Biswas <[email protected]>

server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java

@@ -715,7 +715,8 @@ public static final IndexShard newIndexShard(
             nodeId,
             null,
             DefaultRemoteStoreSettings.INSTANCE,
-            false
+            false,
+            IndexShardTestUtils.getFakeDiscoveryNodes(initializingShardRouting)
         );
     }
 

server/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java

@@ -8,17 +8,29 @@
 
 package org.opensearch.remotemigration;
 
+import org.opensearch.action.DocWriteResponse;
 import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesRequest;
 import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
+import org.opensearch.action.bulk.BulkRequest;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.delete.DeleteResponse;
+import org.opensearch.action.index.IndexRequest;
+import org.opensearch.action.index.IndexResponse;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
+import org.opensearch.common.UUIDs;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.repositories.fs.ReloadableFsRepository;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.nio.file.Path;
+import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING;
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
 import static org.opensearch.repositories.fs.ReloadableFsRepository.REPOSITORIES_FAILRATE_SETTING;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 
@@ -28,8 +40,16 @@ public class MigrationBaseTestCase extends OpenSearchIntegTestCase {
 
     protected Path segmentRepoPath;
     protected Path translogRepoPath;
-
     boolean addRemote = false;
+    Settings extraSettings = Settings.EMPTY;
+
+    private final List<String> documentKeys = List.of(
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5)
+    );
 
     protected Settings nodeSettings(int nodeOrdinal) {
         if (segmentRepoPath == null || translogRepoPath == null) {
@@ -40,6 +60,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
             logger.info("Adding remote store node");
             return Settings.builder()
                 .put(super.nodeSettings(nodeOrdinal))
+                .put(extraSettings)
                 .put(remoteStoreClusterSettings(REPOSITORY_NAME, segmentRepoPath, REPOSITORY_2_NAME, translogRepoPath))
                 .build();
         } else {
@@ -64,4 +85,76 @@ protected void setFailRate(String repoName, int value) throws ExecutionException
             client().admin().cluster().preparePutRepository(repoName).setType(ReloadableFsRepository.TYPE).setSettings(settings).get()
         );
     }
+
+    public void initDocRepToRemoteMigration() {
+        assertTrue(
+            internalCluster().client()
+                .admin()
+                .cluster()
+                .prepareUpdateSettings()
+                .setPersistentSettings(
+                    Settings.builder()
+                        .put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "mixed")
+                        .put(MIGRATION_DIRECTION_SETTING.getKey(), "remote_store")
+                )
+                .get()
+                .isAcknowledged()
+        );
+    }
+
+    public BulkResponse indexBulk(String indexName, int numDocs) {
+        BulkRequest bulkRequest = new BulkRequest();
+        for (int i = 0; i < numDocs; i++) {
+            final IndexRequest request = client().prepareIndex(indexName)
+                .setId(UUIDs.randomBase64UUID())
+                .setSource(documentKeys.get(randomIntBetween(0, documentKeys.size() - 1)), randomAlphaOfLength(5))
+                .request();
+            bulkRequest.add(request);
+        }
+        return client().bulk(bulkRequest).actionGet();
+    }
+
+    private void indexSingleDoc(String indexName) {
+        IndexResponse indexResponse = client().prepareIndex(indexName).setId("id").setSource("field", "value").get();
+        assertEquals(DocWriteResponse.Result.CREATED, indexResponse.getResult());
+        DeleteResponse deleteResponse = client().prepareDelete(indexName, "id").get();
+        assertEquals(DocWriteResponse.Result.DELETED, deleteResponse.getResult());
+        client().prepareIndex(indexName).setSource("auto", true).get();
+    }
+
+    public class AsyncIndexingService {
+        private String indexName;
+        private AtomicLong indexedDocs = new AtomicLong(0);
+        private AtomicBoolean finished = new AtomicBoolean();
+        private Thread indexingThread;
+
+        AsyncIndexingService(String indexName) {
+            this.indexName = indexName;
+        }
+
+        public void startIndexing() {
+            indexingThread = getIndexingThread();
+            indexingThread.start();
+        }
+
+        public void stopIndexing() throws InterruptedException {
+            finished.set(true);
+            indexingThread.join();
+        }
+
+        public long getIndexedDocs() {
+            return indexedDocs.get();
+        }
+
+        private Thread getIndexingThread() {
+            return new Thread(() -> {
+                while (finished.get() == false) {
+                    indexSingleDoc(indexName);
+                    long currentDocCount = indexedDocs.incrementAndGet();
+                    logger.info("Completed ingestion of {} docs", currentDocCount);
+
+                }
+            });
+        }
+    }
 }