Refactor SegmentReplicationTargetService to only hold completed targe…

…t state instead of the entire target. (#11043) (#11081)

(cherry picked from commit 54fa050)

Signed-off-by: Marc Handalian <handalm@amazon.com>
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java

@@ -70,7 +70,7 @@ public class SegmentReplicationTargetService implements IndexEventListener {
 
     private final ReplicationCollection<SegmentReplicationTarget> onGoingReplications;
 
-    private final Map<ShardId, SegmentReplicationTarget> completedReplications = ConcurrentCollections.newConcurrentMap();
+    private final Map<ShardId, SegmentReplicationState> completedReplications = ConcurrentCollections.newConcurrentMap();
 
     private final SegmentReplicationSourceFactory sourceFactory;
 
@@ -192,7 +192,7 @@ public SegmentReplicationState getOngoingEventSegmentReplicationState(ShardId sh
      */
     @Nullable
     public SegmentReplicationState getlatestCompletedEventSegmentReplicationState(ShardId shardId) {
-        return Optional.ofNullable(completedReplications.get(shardId)).map(SegmentReplicationTarget::state).orElse(null);
+        return completedReplications.get(shardId);
     }
 
     /**
@@ -525,7 +525,7 @@ public void onResponse(Void o) {
                 logger.debug(() -> new ParameterizedMessage("Finished replicating {} marking as done.", target.description()));
                 onGoingReplications.markAsDone(replicationId);
                 if (target.state().getIndex().recoveredFileCount() != 0 && target.state().getIndex().recoveredBytes() != 0) {
-                    completedReplications.put(target.shardId(), target);
+                    completedReplications.put(target.shardId(), target.state());
                 }
             }