Fix the replication failure issues

Signed-off-by: Sachin Kale <[email protected]>
opensearch-project · Apr 9, 2024 · d605254 · d605254
1 parent 84557bc
commit d605254
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 7 deletions.
diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -5054,7 +5054,11 @@ public void syncSegmentsFromRemoteSegmentStore(boolean overrideLocal, final Runn
                         .filter(file -> file.startsWith(RemoteSegmentStoreDirectory.SEGMENT_INFOS_SNAPSHOT_PREFIX))
                         .collect(Collectors.toList());
                     assert segmentInfosSnapshotFilenames.size() == 1;
-                    infosSnapshot = SegmentInfos.readCommit(store.directory(), segmentInfosSnapshotFilenames.get(0));
+                    infosSnapshot = SegmentInfos.readCommit(
+                        store.directory(),
+                        store.directory().openChecksumInput(segmentInfosSnapshotFilenames.get(0), IOContext.READ),
+                        remoteSegmentMetadata.getGeneration()
+                    );
                 } else {
                     infosSnapshot = store.buildSegmentInfos(
                         remoteSegmentMetadata.getSegmentInfosBytes(),

diff --git a/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java b/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java
@@ -652,7 +652,12 @@ public void uploadMetadata(
                             ) {
                                 segmentInfosSnapshot.write(segmentInfosIndexOutput);
                             }
-                            copyFrom(storeDirectory, segmentInfoSnapshotFilename, segmentInfoSnapshotFilename, IOContext.DEFAULT);
+                            remoteDataDirectory.copyFrom(
+                                storeDirectory,
+                                segmentInfoSnapshotFilename,
+                                segmentInfoSnapshotFilename,
+                                IOContext.DEFAULT
+                            );
                             String segmentInfosSnapshotChecksum = getChecksumOfLocalFile(storeDirectory, segmentInfoSnapshotFilename);
                             UploadedSegmentMetadata segmentInfosSnapshotMetadata = new UploadedSegmentMetadata(
                                 segmentInfoSnapshotFilename,

diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java
@@ -25,6 +25,7 @@
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.index.shard.IndexShard;
+import org.opensearch.index.store.RemoteSegmentStoreDirectory;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.StoreFileMetadata;
 import org.opensearch.indices.recovery.MultiFileWriter;
@@ -36,6 +37,7 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
@@ -283,19 +285,31 @@ private void updateFileRecoveryBytes(String fileName, long bytesRecovered) {
     private void finalizeReplication(CheckpointInfoResponse checkpointInfoResponse) throws OpenSearchCorruptionException {
         cancellableThreads.checkForCancel();
         state.setStage(SegmentReplicationState.Stage.FINALIZE_REPLICATION);
+        byte[] segmentInfosBytes = checkpointInfoResponse.getInfosBytes();
         // Handle empty SegmentInfos bytes for recovering replicas
-        if (checkpointInfoResponse.getInfosBytes() == null) {
+        if (segmentInfosBytes == null) {
             return;
         }
         Store store = null;
         try {
             store = store();
             store.incRef();
             multiFileWriter.renameAllTempFiles();
-            final SegmentInfos infos = store.buildSegmentInfos(
-                checkpointInfoResponse.getInfosBytes(),
-                checkpointInfoResponse.getCheckpoint().getSegmentsGen()
-            );
+            final SegmentInfos infos;
+            if (segmentInfosBytes.length == 0) {
+                List<String> segmentInfosSnapshotFilenames = Arrays.stream(store.directory().listAll())
+                    .filter(file -> file.startsWith(RemoteSegmentStoreDirectory.SEGMENT_INFOS_SNAPSHOT_PREFIX))
+                    .collect(Collectors.toList());
+                assert segmentInfosSnapshotFilenames.size() == 1;
+                infos = SegmentInfos.readCommit(
+                    store.directory(),
+                    store.directory().openChecksumInput(segmentInfosSnapshotFilenames.get(0), IOContext.READ),
+                    checkpointInfoResponse.getCheckpoint().getSegmentsGen()
+                );
+                store.deleteQuiet(segmentInfosSnapshotFilenames.get(0));
+            } else {
+                infos = store.buildSegmentInfos(segmentInfosBytes, checkpointInfoResponse.getCheckpoint().getSegmentsGen());
+            }
             indexShard.finalizeReplication(infos);
         } catch (CorruptIndexException | IndexFormatTooNewException | IndexFormatTooOldException ex) {
             // this is a fatal exception at this stage.