Address review: rename async lease method, return early during force resume bootstrap

Signed-off-by: Mohit Kumar <mohitamg@amazon.com>

Author: mohitamg

src/main/kotlin/org/opensearch/replication/action/resume/ForceResumeCoordinator.kt

@@ -121,7 +121,7 @@ class ForceResumeCoordinator(
             val retainingSeqNo = getLeaderGlobalCheckpoint(remoteClient, params.leaderIndex.name, leaderShardId.id) + 1
 
             try {
-                retentionLeaseHelper.addRetentionLease(
+                retentionLeaseHelper.addRetentionLeaseAsync(
                     leaderShardId, retainingSeqNo, followerShardId
                 )
             } catch (e: RetentionLeaseAlreadyExistsException) {

src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt

@@ -140,10 +140,18 @@ class TransportResumeIndexReplicationAction @Inject constructor(transportService
                     listener.onResponse(ReplicateIndexResponse(false))
                 }
 
-                // Now wait for the replication to start and the follower index to get created before returning
+                // Wait for the replication task to start. For force resume (snapshot bootstrap),
+                // return once RESTORING state is reached — same as start replication — since
+                // bootstrap can take a long time for large indices. For normal resume, wait
+                // until FOLLOWING since no snapshot restore is needed.
                 persistentTasksService.waitForTaskCondition(task.id, request.timeout()) { t ->
                     val replicationState = (t.state as IndexReplicationState?)?.state
-                    replicationState == ReplicationState.FOLLOWING
+                    if (request.forceResume) {
+                        replicationState == ReplicationState.FOLLOWING ||
+                                replicationState == ReplicationState.RESTORING
+                    } else {
+                        replicationState == ReplicationState.FOLLOWING
+                    }
                 }
 
                 AcknowledgedResponse(true)

src/main/kotlin/org/opensearch/replication/seqno/RemoteClusterRetentionLeaseHelper.kt

@@ -116,11 +116,13 @@ class RemoteClusterRetentionLeaseHelper constructor(var followerClusterNameWithU
     }
 
     /**
-     * Suspend (non-blocking) version of addRetentionLease for use in coroutine contexts
-     * (e.g., transport thread). The blocking overload below uses actionGet() and must NOT
-     * be called from transport threads.
+     * Non-blocking (suspend) version of addRetentionLease for use in coroutine contexts.
+     * The blocking overload [addRetentionLease(ShardId, Long, ShardId, Long)] uses actionGet()
+     * and is intended for synchronous callers (e.g., leader-side restore service).
+     * This suspend variant must NOT be called from transport threads directly — only from
+     * coroutine scopes (e.g., ForceResumeCoordinator).
      */
-    public suspend fun addRetentionLease(leaderShardId: ShardId, seqNo: Long, followerShardId: ShardId) {
+    public suspend fun addRetentionLeaseAsync(leaderShardId: ShardId, seqNo: Long, followerShardId: ShardId) {
         val retentionLeaseId = retentionLeaseIdForShard(followerClusterNameWithUUID, followerShardId)
         val request = RetentionLeaseActions.AddRequest(leaderShardId, retentionLeaseId, seqNo, retentionLeaseSource)
         log.info("Adding retention lease $retentionLeaseId (async)")