Avoid use of indicesService in Resume replicaiton flow (#1030) (#1036)

Avoid use of indicesService in Resume replicaiton flow. Signed-off-by: monusingh-1 <[email protected]> (cherry picked from commit 66d126e) Co-authored-by: Monu Singh <[email protected]>
opensearch-project · Jul 6, 2023 · 4656df7 · 4656df7
1 parent 6c2e108
commit 4656df7
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 7 deletions.
diff --git a/.../kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt b/.../kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt
@@ -54,7 +54,6 @@ import org.opensearch.common.io.stream.StreamInput
 import org.opensearch.env.Environment
 import org.opensearch.index.IndexNotFoundException
 import org.opensearch.index.shard.ShardId
-import org.opensearch.replication.util.indicesService
 import org.opensearch.threadpool.ThreadPool
 import org.opensearch.transport.TransportService
 import java.io.IOException
@@ -136,10 +135,7 @@ class TransportResumeIndexReplicationAction @Inject constructor(transportService
         shards.forEach {
             val followerShardId = it.value.shardId
 
-            val followerIndexService = indicesService.indexServiceSafe(followerShardId.index)
-            val indexShard = followerIndexService.getShard(followerShardId.id)
-
-            if  (!retentionLeaseHelper.verifyRetentionLeaseExist(ShardId(params.leaderIndex, followerShardId.id), followerShardId, indexShard.lastSyncedGlobalCheckpoint+1)) {
+            if  (!retentionLeaseHelper.verifyRetentionLeaseExist(ShardId(params.leaderIndex, followerShardId.id), followerShardId)) {
                 isResumable = false
             }
         }

diff --git a/src/main/kotlin/org/opensearch/replication/seqno/RemoteClusterRetentionLeaseHelper.kt b/src/main/kotlin/org/opensearch/replication/seqno/RemoteClusterRetentionLeaseHelper.kt
@@ -57,7 +57,7 @@ class RemoteClusterRetentionLeaseHelper constructor(var followerClusterNameWithU
         }
     }
 
-    public suspend fun verifyRetentionLeaseExist(leaderShardId: ShardId, followerShardId: ShardId, seqNo: Long): Boolean  {
+    public suspend fun verifyRetentionLeaseExist(leaderShardId: ShardId, followerShardId: ShardId): Boolean  {
         val retentionLeaseId = retentionLeaseIdForShard(followerClusterNameWithUUID, followerShardId)
         // Currently there is no API to describe/list the retention leases .
         // So we are verifying the existence of lease by trying to renew a lease by same name .
@@ -73,7 +73,7 @@ class RemoteClusterRetentionLeaseHelper constructor(var followerClusterNameWithU
             return true
         }
         catch (e: RetentionLeaseNotFoundException) {
-            return addNewRetentionLeaseIfOldExists(leaderShardId, followerShardId, seqNo)
+            return addNewRetentionLeaseIfOldExists(leaderShardId, followerShardId, RetentionLeaseActions.RETAIN_ALL)
         }catch (e : Exception) {
             return false
         }