[Backport 2.x] [Segment Replication] Moving RecoveryState.Index to a …

…top-level class and renaming (#3971) * Backport #3075 to 2.x branch and resolve conflict in RecoveryState class Signed-off-by: Rishikesh1159 <[email protected]> Co-authored-by: Kartik Ganesh <[email protected]>
opensearch-project · Jul 22, 2022 · c0b1b59 · c0b1b59
1 parent 0a42e6d
commit c0b1b59
Show file tree

Hide file tree

Showing 14 changed files with 639 additions and 574 deletions.
diff --git a/server/src/internalClusterTest/java/org/opensearch/gateway/RecoveryFromGatewayIT.java b/server/src/internalClusterTest/java/org/opensearch/gateway/RecoveryFromGatewayIT.java
@@ -62,6 +62,7 @@
 import org.opensearch.index.shard.ShardPath;
 import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.recovery.RecoveryState;
+import org.opensearch.indices.replication.common.ReplicationLuceneIndex;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.test.OpenSearchIntegTestCase;
 import org.opensearch.test.OpenSearchIntegTestCase.ClusterScope;
@@ -547,7 +548,7 @@ public void testReuseInFileBasedPeerRecovery() throws Exception {
         final Set<String> files = new HashSet<>();
         for (final RecoveryState recoveryState : initialRecoveryReponse.shardRecoveryStates().get("test")) {
             if (recoveryState.getTargetNode().getName().equals(replicaNode)) {
-                for (final RecoveryState.FileDetail file : recoveryState.getIndex().fileDetails()) {
+                for (final ReplicationLuceneIndex.FileMetadata file : recoveryState.getIndex().fileDetails()) {
                     files.add(file.name());
                 }
                 break;
@@ -607,7 +608,7 @@ public Settings onNodeStopped(String nodeName) throws Exception {
             long reused = 0;
             int filesRecovered = 0;
             int filesReused = 0;
-            for (final RecoveryState.FileDetail file : recoveryState.getIndex().fileDetails()) {
+            for (final ReplicationLuceneIndex.FileMetadata file : recoveryState.getIndex().fileDetails()) {
                 if (files.contains(file.name()) == false) {
                     recovered += file.length();
                     filesRecovered++;

diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/recovery/IndexRecoveryIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/recovery/IndexRecoveryIT.java
@@ -102,6 +102,7 @@
 import org.opensearch.indices.NodeIndicesStats;
 import org.opensearch.indices.analysis.AnalysisModule;
 import org.opensearch.indices.recovery.RecoveryState.Stage;
+import org.opensearch.indices.replication.common.ReplicationLuceneIndex;
 import org.opensearch.node.NodeClosedException;
 import org.opensearch.node.RecoverySettingsChunkSizePlugin;
 import org.opensearch.plugins.AnalysisPlugin;
@@ -836,7 +837,7 @@ private IndicesStatsResponse createAndPopulateIndex(String name, int nodeCount,
         return client().admin().indices().prepareStats(name).execute().actionGet();
     }
 
-    private void validateIndexRecoveryState(RecoveryState.Index indexState) {
+    private void validateIndexRecoveryState(ReplicationLuceneIndex indexState) {
         assertThat(indexState.time(), greaterThanOrEqualTo(0L));
         assertThat(indexState.recoveredFilesPercent(), greaterThanOrEqualTo(0.0f));
         assertThat(indexState.recoveredFilesPercent(), lessThanOrEqualTo(100.0f));

diff --git a/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java b/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java
@@ -62,6 +62,7 @@
 import org.opensearch.index.store.Store;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.indices.recovery.RecoveryState;
+import org.opensearch.indices.replication.common.ReplicationLuceneIndex;
 import org.opensearch.repositories.IndexId;
 import org.opensearch.repositories.Repository;
 
@@ -178,7 +179,7 @@ void recoverFromLocalShards(
     }
 
     void addIndices(
-        final RecoveryState.Index indexRecoveryStats,
+        final ReplicationLuceneIndex indexRecoveryStats,
         final Directory target,
         final Sort indexSort,
         final Directory[] sources,
@@ -235,9 +236,9 @@ void addIndices(
      * @opensearch.internal
      */
     static final class StatsDirectoryWrapper extends FilterDirectory {
-        private final RecoveryState.Index index;
+        private final ReplicationLuceneIndex index;
 
-        StatsDirectoryWrapper(Directory in, RecoveryState.Index indexRecoveryStats) {
+        StatsDirectoryWrapper(Directory in, ReplicationLuceneIndex indexRecoveryStats) {
             super(in);
             this.index = indexRecoveryStats;
         }
@@ -358,7 +359,7 @@ private ActionListener<Boolean> recoveryListener(IndexShard indexShard, ActionLi
                     + "]";
 
                 if (logger.isTraceEnabled()) {
-                    RecoveryState.Index index = recoveryState.getIndex();
+                    ReplicationLuceneIndex index = recoveryState.getIndex();
                     StringBuilder sb = new StringBuilder();
                     sb.append("    index    : files           [")
                         .append(index.totalFileCount())
@@ -475,7 +476,7 @@ private void internalRecoverFromStore(IndexShard indexShard) throws IndexShardRe
                     writeEmptyRetentionLeasesFile(indexShard);
                 }
                 // since we recover from local, just fill the files and size
-                final RecoveryState.Index index = recoveryState.getIndex();
+                final ReplicationLuceneIndex index = recoveryState.getIndex();
                 try {
                     if (si != null) {
                         addRecoveredFileDetails(si, store, index);
@@ -513,7 +514,7 @@ private static void writeEmptyRetentionLeasesFile(IndexShard indexShard) throws
         assert indexShard.loadRetentionLeases().leases().isEmpty();
     }
 
-    private void addRecoveredFileDetails(SegmentInfos si, Store store, RecoveryState.Index index) throws IOException {
+    private void addRecoveredFileDetails(SegmentInfos si, Store store, ReplicationLuceneIndex index) throws IOException {
         final Directory directory = store.directory();
         for (String name : Lucene.files(si)) {
             long length = directory.fileLength(name);

diff --git a/server/src/main/java/org/opensearch/indices/recovery/MultiFileWriter.java b/server/src/main/java/org/opensearch/indices/recovery/MultiFileWriter.java
@@ -44,6 +44,7 @@
 import org.opensearch.common.util.concurrent.ConcurrentCollections;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.StoreFileMetadata;
+import org.opensearch.indices.replication.common.ReplicationLuceneIndex;
 import org.opensearch.transport.Transports;
 
 import java.io.IOException;
@@ -63,7 +64,7 @@
  */
 public class MultiFileWriter extends AbstractRefCounted implements Releasable {
 
-    public MultiFileWriter(Store store, RecoveryState.Index indexState, String tempFilePrefix, Logger logger, Runnable ensureOpen) {
+    public MultiFileWriter(Store store, ReplicationLuceneIndex indexState, String tempFilePrefix, Logger logger, Runnable ensureOpen) {
         super("multi_file_writer");
         this.store = store;
         this.indexState = indexState;
@@ -76,7 +77,7 @@ public MultiFileWriter(Store store, RecoveryState.Index indexState, String tempF
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final Logger logger;
     private final Store store;
-    private final RecoveryState.Index indexState;
+    private final ReplicationLuceneIndex indexState;
     private final String tempFilePrefix;
 
     private final ConcurrentMap<String, IndexOutput> openIndexOutputs = ConcurrentCollections.newConcurrentMap();

diff --git a/server/src/main/java/org/opensearch/indices/recovery/PeerRecoveryTargetService.java b/server/src/main/java/org/opensearch/indices/recovery/PeerRecoveryTargetService.java
@@ -70,6 +70,7 @@
 import org.opensearch.index.translog.Translog;
 import org.opensearch.index.translog.TranslogCorruptedException;
 import org.opensearch.indices.recovery.RecoveriesCollection.RecoveryRef;
+import org.opensearch.indices.replication.common.ReplicationLuceneIndex;
 import org.opensearch.indices.replication.common.ReplicationTimer;
 import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
@@ -533,8 +534,8 @@ public void messageReceived(final RecoveryFileChunkRequest request, TransportCha
                     return;
                 }
 
-                final RecoveryState.Index indexState = recoveryTarget.state().getIndex();
-                if (request.sourceThrottleTimeInNanos() != RecoveryState.Index.UNKNOWN) {
+                final ReplicationLuceneIndex indexState = recoveryTarget.state().getIndex();
+                if (request.sourceThrottleTimeInNanos() != ReplicationLuceneIndex.UNKNOWN) {
                     indexState.addSourceThrottling(request.sourceThrottleTimeInNanos());
                 }