Fix flaky test SegmentReplicationTargetServiceTests.testStartReplicationListenerSuccess (#17969)

* fix flaky test SegmentReplicationTargetServiceTests.testStartReplicationListenerSuccess

Signed-off-by: guojialiang <[email protected]>

* add UT

Signed-off-by: guojialiang <[email protected]>

* add comment

Signed-off-by: guojialiang <[email protected]>

* add tests

Signed-off-by: guojialiang <[email protected]>

---------

Signed-off-by: guojialiang <[email protected]>

Author: guojialiang92

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -1716,16 +1716,24 @@ public void releaseLockOnCommitData(String snapshotId, long primaryTerm, long ge
     }
 
     public Optional<NRTReplicationEngine> getReplicationEngine() {
-        if (getEngine() instanceof NRTReplicationEngine) {
-            return Optional.of((NRTReplicationEngine) getEngine());
-        } else {
+        try {
+            if (getEngine() instanceof NRTReplicationEngine) {
+                return Optional.of((NRTReplicationEngine) getEngine());
+            } else {
+                return Optional.empty();
+            }
+        } catch (AlreadyClosedException e) {
+            // If shard already closed, return empty. The logic related to segment replication will not continue to execute after judging
+            // that the return value is empty, and there will be no side effects.
+            logger.debug("failed to get ReplicationEngine", e);
             return Optional.empty();
         }
     }
 
     public void finalizeReplication(SegmentInfos infos) throws IOException {
-        if (getReplicationEngine().isPresent()) {
-            getReplicationEngine().get().updateSegments(infos);
+        Optional<NRTReplicationEngine> engineOptional = getReplicationEngine();
+        if (engineOptional.isPresent()) {
+            engineOptional.get().updateSegments(infos);
         }
     }
 

server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java

@@ -146,6 +146,10 @@ public void testReplication() throws Exception {
 
             // Assertions
             shards.assertAllEqual(numDocs);
+            assertTrue(primaryShard.getReplicationEngine().isEmpty());
+            assertFalse(replicaShard.getReplicationEngine().isEmpty());
+            replicaShard.close("test", false, false);
+            assertTrue(replicaShard.getReplicationEngine().isEmpty());
         }
     }