addressing comments + fix gradle check

Signed-off-by: Poojita Raj <poojiraj@amazon.com>
opensearch-project · Jun 13, 2022 · ebb2b5e · ebb2b5e
1 parent f366918
commit ebb2b5e
Show file tree

Hide file tree

Showing 6 changed files with 28 additions and 4 deletions.
diff --git a/server/src/main/java/org/opensearch/OpenSearchException.java b/server/src/main/java/org/opensearch/OpenSearchException.java
@@ -1594,6 +1594,12 @@ private enum OpenSearchExceptionHandle {
             org.opensearch.transport.NoSeedNodeLeftException::new,
             160,
             LegacyESVersion.V_7_10_0
+        ),
+        REPLICATION_FAILED_EXCEPTION(
+            org.opensearch.indices.replication.common.ReplicationFailedException.class,
+            org.opensearch.indices.replication.common.ReplicationFailedException::new,
+            161,
+            UNKNOWN_VERSION_ADDED
         );
 
         final Class<? extends OpenSearchException> exceptionClass;

diff --git a/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java b/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java
@@ -85,7 +85,6 @@ public NRTReplicationEngine(EngineConfig engineConfig) {
 
     public synchronized void updateSegments(final SegmentInfos infos, long seqNo) throws IOException {
         // Update the current infos reference on the Engine's reader.
-        assert engineConfig.isReadOnlyReplica() : "Only replicas should update Infos";
         readerManager.updateSegments(infos);
 
         // only update the persistedSeqNo and "lastCommitted" infos reference if the incoming segments have a higher
@@ -95,7 +94,6 @@ public synchronized void updateSegments(final SegmentInfos infos, long seqNo) th
             rollTranslogGeneration();
         }
         localCheckpointTracker.fastForwardProcessedSeqNo(seqNo);
-        readerManager.maybeRefresh();
     }
 
     @Override

diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -1364,9 +1364,18 @@ public GatedCloseable<IndexCommit> acquireLastIndexCommit(boolean flushFirst) th
         }
     }
 
+    public Optional<NRTReplicationEngine> getReplicationEngine() {
+        if (getEngine() instanceof NRTReplicationEngine) {
+            return Optional.of((NRTReplicationEngine) getEngine());
+        } else {
+            return Optional.empty();
+        }
+    }
+
     public void finalizeReplication(SegmentInfos infos, long seqNo) throws IOException {
-        assert getEngine() instanceof NRTReplicationEngine;
-        ((NRTReplicationEngine) getEngine()).updateSegments(infos, seqNo);
+        if (getReplicationEngine().isPresent()) {
+            getReplicationEngine().get().updateSegments(infos, seqNo);
+        }
     }
 
     /**

diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java
@@ -157,6 +157,10 @@ private void getFiles(CheckpointInfoResponse checkpointInfo, StepListener<GetSeg
         Store.MetadataSnapshot localMetadata = getMetadataSnapshot();
         final Store.RecoveryDiff diff = snapshot.recoveryDiff(localMetadata);
         logger.debug("Replication diff {}", diff);
+        // Segments are immutable. So if the replica has any segments with the same name that differ from the one in the incoming snapshot
+        // from
+        // source that means the local copy of the segment has been corrupted/changed in some way and we throw an IllegalStateException to
+        // fail the shard
         if (diff.different.isEmpty() == false) {
             getFilesListener.onFailure(
                 new IllegalStateException(

diff --git a/...r/src/main/java/org/opensearch/indices/replication/common/ReplicationFailedException.java b/...r/src/main/java/org/opensearch/indices/replication/common/ReplicationFailedException.java
@@ -40,6 +40,11 @@
 
 import java.io.IOException;
 
+/**
+ * Exception thrown if replication fails
+ *
+ * @opensearch.internal
+ */
 public class ReplicationFailedException extends OpenSearchException {
 
     public ReplicationFailedException(IndexShard shard, Throwable cause) {

diff --git a/server/src/test/java/org/opensearch/ExceptionSerializationTests.java b/server/src/test/java/org/opensearch/ExceptionSerializationTests.java
@@ -84,6 +84,7 @@
 import org.opensearch.indices.InvalidIndexTemplateException;
 import org.opensearch.indices.recovery.PeerRecoveryNotFound;
 import org.opensearch.indices.recovery.RecoverFilesRecoveryException;
+import org.opensearch.indices.replication.common.ReplicationFailedException;
 import org.opensearch.ingest.IngestProcessorException;
 import org.opensearch.cluster.coordination.NodeHealthCheckFailureException;
 import org.opensearch.repositories.RepositoryException;
@@ -849,6 +850,7 @@ public void testIds() {
         ids.put(158, PeerRecoveryNotFound.class);
         ids.put(159, NodeHealthCheckFailureException.class);
         ids.put(160, NoSeedNodeLeftException.class);
+        ids.put(161, ReplicationFailedException.class);
 
         Map<Class<? extends OpenSearchException>, Integer> reverse = new HashMap<>();
         for (Map.Entry<Integer, Class<? extends OpenSearchException>> entry : ids.entrySet()) {