addressing comments + fix gradle check

Poojita-Raj · Poojita-Raj · commit ebb2b5ea0f55 · 2022-06-13T13:06:20.000-07:00
Signed-off-by: Poojita Raj &lt;poojiraj@amazon.com&gt;
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/server/src/main/java/org/opensearch/indices/replication/common/ReplicationFailedException.java b/server/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()) {

Original file line number	Diff line number	Diff line change
`@@ -85,7 +85,6 @@ public NRTReplicationEngine(EngineConfig engineConfig) {`
`85`	`85`
`86`	`86`	`public synchronized void updateSegments(final SegmentInfos infos, long seqNo) throws IOException {`
`87`	`87`	`// Update the current infos reference on the Engine's reader.`
`88`		`- assert engineConfig.isReadOnlyReplica() : "Only replicas should update Infos";`
`89`	`88`	`readerManager.updateSegments(infos);`
`90`	`89`
`91`	`90`	`// 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`
`95`	`94`	`rollTranslogGeneration();`
`96`	`95`	`}`
`97`	`96`	`localCheckpointTracker.fastForwardProcessedSeqNo(seqNo);`
`98`		`- readerManager.maybeRefresh();`
`99`	`97`	`}`
`100`	`98`
`101`	`99`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -1364,9 +1364,18 @@ public GatedCloseable<IndexCommit> acquireLastIndexCommit(boolean flushFirst) th`
`1364`	`1364`	`}`
`1365`	`1365`	`}`
`1366`	`1366`
	`1367`	`+ public Optional<NRTReplicationEngine> getReplicationEngine() {`
	`1368`	`+ if (getEngine() instanceof NRTReplicationEngine) {`
	`1369`	`+ return Optional.of((NRTReplicationEngine) getEngine());`
	`1370`	`+ } else {`
	`1371`	`+ return Optional.empty();`
	`1372`	`+ }`
	`1373`	`+ }`
	`1374`	`+`
`1367`	`1375`	`public void finalizeReplication(SegmentInfos infos, long seqNo) throws IOException {`
`1368`		`- assert getEngine() instanceof NRTReplicationEngine;`
`1369`		`- ((NRTReplicationEngine) getEngine()).updateSegments(infos, seqNo);`
	`1376`	`+ if (getReplicationEngine().isPresent()) {`
	`1377`	`+ getReplicationEngine().get().updateSegments(infos, seqNo);`
	`1378`	`+ }`
`1370`	`1379`	`}`
`1371`	`1380`
`1372`	`1381`	`/**`