Add force resume support for cross-cluster replication by mohit10011999 · Pull Request #1685 · opensearch-project/cross-cluster-replication

mohit10011999 · 2026-05-09T13:00:51Z

Description

When retention leases expire on the leader cluster (e.g., due to prolonged pause or leader translog rollover), the normal resume API fails because it cannot catch up from where it left off. This PR adds a force_resume option that triggers a snapshot-based bootstrap to recover replication.

Solution

Added force_resume=true parameter to the resume replication API:

POST /_plugins/_replication/<follower-index>/_resume
{ "force_resume": true }

Backward Compatibility

The force_resume field defaults to false, so existing clients calling _resume with an empty body or without the field see no behavior change
Wire serialization is additive (new boolean field appended)

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

New functionality includes testing.
New functionality has been documented.
API changes companion pull request created.
Commits are signed per the DCO using --signoff.
Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

github-actions · 2026-05-10T08:46:46Z

PR Reviewer Guide 🔍

(Review updated until commit `82ea354`)

Here are some key observations to aid the review process:

🧪 PR contains tests
🔒 No security concerns identified
✅ No TODO sections
🔀 No multiple PR themes
⚡ Recommended focus areas for review Race Condition After deleting the follower index in `executeForceResume`, the code checks `clusterService.state().routingTable.hasIndex(followerIndexName)` before deletion. However, cluster state updates are asynchronous. If the index deletion completes but the routing table hasn't updated yet when the subsequent start replication call executes, the start operation may fail or behave unexpectedly because it expects the index to not exist. This can occur under high load or slow cluster state propagation. if (clusterService.state().routingTable.hasIndex(followerIndexName)) { client.suspending(client.admin().indices()::delete, defaultContext = true)( org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) ) log.info("Force resume: deleted standalone follower index $followerIndexName") } Incomplete Cleanup The `executeForceResume` function performs stop and delete operations but does not verify that these operations completed successfully before proceeding to start replication. If the stop operation fails partway (e.g., task cleanup fails but metadata is removed), or if the delete operation fails, the subsequent start operation will encounter an inconsistent state. This can lead to replication failure or orphaned resources. client.suspendExecute(STOP_REPLICATION_ACTION_TYPE, StopIndexReplicationRequest(followerIndexName), defaultContext = true) log.info("Force resume: replication stopped for $followerIndexName") // Step 2: Delete the follower index so start replication can create it fresh. // After stop, the follower is a standalone index. Start replication requires // the follower index to NOT exist (it creates it via snapshot restore). if (clusterService.state().routingTable.hasIndex(followerIndexName)) { client.suspending(client.admin().indices()::delete, defaultContext = true)( org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) ) log.info("Force resume: deleted standalone follower index $followerIndexName") } // Step 3: Full start — fresh metadata, task creation, snapshot restore. // The RemoteClusterRepository restore path acquires retention leases during // the snapshot restore process, so no separate lease acquisition is needed. val startRequest = ReplicateIndexRequest(followerIndexName, connectionName, leaderIndexName, settings) // Skip setup checks since this is an internal re-start (user already had permissions // when replication was originally set up). Same pattern as autofollow. startRequest.isAutoFollowRequest = true if (followerUser != null && leaderUser != null) { startRequest.useRoles = hashMapOf( ReplicateIndexRequest.FOLLOWER_CLUSTER_ROLE to followerUser.name, ReplicateIndexRequest.LEADER_CLUSTER_ROLE to leaderUser.name ) } client.suspendExecute(ReplicateIndexAction.INSTANCE, startRequest, defaultContext = true) log.info("Force resume: replication restarted for $followerIndexName")

github-actions · 2026-05-10T08:47:24Z

PR Code Suggestions ✨

Latest suggestions up to 82ea354

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Add error handling for index deletion Add error handling for the index deletion operation. If deletion fails (e.g., due to concurrent operations or cluster state issues), the force resume will proceed to start replication, which will fail because the index still exists. Wrap the deletion in a try-catch block and handle failures appropriately. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [173-178] if (clusterService.state().routingTable.hasIndex(followerIndexName)) { - client.suspending(client.admin().indices()::delete, defaultContext = true)( - org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) - ) - log.info("Force resume: deleted standalone follower index $followerIndexName") + try { + client.suspending(client.admin().indices()::delete, defaultContext = true)( + org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) + ) + log.info("Force resume: deleted standalone follower index $followerIndexName") + } catch (e: Exception) { + log.error("Force resume: failed to delete follower index $followerIndexName", e) + throw IllegalStateException("Failed to delete follower index during force resume: ${e.message}", e) + } } Suggestion importance[1-10]: 7 __ Why: Adding error handling for index deletion is important to prevent silent failures and provide clear error messages. However, the deletion operation is already within a suspend function that will propagate exceptions, so the try-catch mainly adds explicit logging and error context rather than fixing a critical bug.	Medium
	Add error handling for stop operation Add error handling for the stop replication operation. If the stop operation fails, the subsequent delete and start operations will execute on an index that may still have active replication tasks or metadata, leading to inconsistent state. Wrap in try-catch and propagate the error. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [167-168] -client.suspendExecute(STOP_REPLICATION_ACTION_TYPE, StopIndexReplicationRequest(followerIndexName), defaultContext = true) -log.info("Force resume: replication stopped for $followerIndexName") +try { + client.suspendExecute(STOP_REPLICATION_ACTION_TYPE, StopIndexReplicationRequest(followerIndexName), defaultContext = true) + log.info("Force resume: replication stopped for $followerIndexName") +} catch (e: Exception) { + log.error("Force resume: failed to stop replication for $followerIndexName", e) + throw IllegalStateException("Failed to stop replication during force resume: ${e.message}", e) +} Suggestion importance[1-10]: 7 __ Why: Adding error handling for the stop operation improves error visibility and helps diagnose failures. However, the `suspendExecute` call will already propagate exceptions up the call stack, so this is more about adding explicit error logging and context rather than preventing a critical failure.	Medium
	Add error handling for start operation Add error handling for the start replication operation. If the start operation fails after stop and delete have succeeded, the index will be left in a deleted state with no replication. Wrap in try-catch to provide clear error messaging to the user about the failure state. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [193-194] -client.suspendExecute(ReplicateIndexAction.INSTANCE, startRequest, defaultContext = true) -log.info("Force resume: replication restarted for $followerIndexName") +try { + client.suspendExecute(ReplicateIndexAction.INSTANCE, startRequest, defaultContext = true) + log.info("Force resume: replication restarted for $followerIndexName") +} catch (e: Exception) { + log.error("Force resume: failed to restart replication for $followerIndexName after stop+delete", e) + throw IllegalStateException("Failed to restart replication during force resume. Index may be deleted: ${e.message}", e) +} Suggestion importance[1-10]: 7 __ Why: Adding error handling for the start operation provides better error messaging about the failure state. However, the `suspendExecute` call will already propagate exceptions, and the `completeWith` block at line 83 will handle exceptions. This suggestion mainly improves error context rather than fixing a critical issue.	Medium

Previous suggestions

Suggestions up to commit 2bf84b4

Category Suggestion Impact

Possible issue

Prevent data loss during force resume

Deleting the follower index during force resume causes permanent data loss of any
documents that were successfully replicated before the pause. If the force resume
fails after deletion (e.g., network issues, leader unavailable), the user loses all
replicated data with no recovery path. Consider preserving the index with a
temporary rename or implementing a rollback mechanism.

src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [170-178]

-// Step 2: Delete the follower index so start replication can create it fresh.
-// After stop, the follower is a standalone index. Start replication requires
-// the follower index to NOT exist (it creates it via snapshot restore).
+// Step 2: Rename the follower index as backup before deletion
+// This allows rollback if force resume fails after this point
+val backupIndexName = "${followerIndexName}_force_resume_backup_${System.currentTimeMillis()}"
 if (clusterService.state().routingTable.hasIndex(followerIndexName)) {
-    client.suspending(client.admin().indices()::delete, defaultContext = true)(
-        org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName)
+    client.suspending(client.admin().indices()::close, defaultContext = true)(
+        org.opensearch.action.admin.indices.close.CloseIndexRequest(followerIndexName)
     )
-    log.info("Force resume: deleted standalone follower index $followerIndexName")
+    // Rename instead of delete to preserve data for potential rollback
+    // User can manually delete backup after verifying force resume success
+    log.warn("Force resume: renaming $followerIndexName to $backupIndexName. Delete backup manually after verification.")
 }

Suggestion importance[1-10]: 7

__

Why: Valid concern about data loss during force resume. The suggestion to rename instead of delete provides a safety mechanism. However, the improved code introduces complexity (close operation, timestamp-based naming) and doesn't fully address the scenario where the renamed index accumulates over multiple failed attempts.

Medium

Verify force resume success before returning

The force resume path returns immediately after executeForceResume without verifying
that replication actually started successfully. If the internal stop or start
operations fail silently or partially, the user receives a success response despite
replication not being active. Add error handling and verification that replication
reached an active state before returning success.

src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [102-108]

 if (!resumable && request.forceResume) {
     // Force resume via internal stop + start.
     // This reuses the battle-tested stop and start paths, ensuring proper
     // cleanup of metadata, checkpoints, tasks, and index blocks.
-    executeForceResume(request.indexName, replMetadata, remoteMetadata)
+    try {
+        executeForceResume(request.indexName, replMetadata, remoteMetadata)
+        // Verify replication started successfully
+        replicationMetadataManager.waitForReplicationState(request.indexName, 
+            ReplicationOverallState.RUNNING, TimeValue.timeValueSeconds(30))
+    } catch (e: Exception) {
+        log.error("Force resume failed for ${request.indexName}", e)
+        throw e
+    }
     return@completeWith AcknowledgedResponse(true)
 }

Suggestion importance[1-10]: 6

__

Why: The suggestion correctly identifies that executeForceResume returns immediately without verification. However, the improved code references a non-existent waitForReplicationState method and assumes synchronous verification is possible in this context. The suggestion is valid in principle but the implementation is questionable.

Low

Suggestions up to commit 576641a

Category	Suggestion	Impact
Possible issue	Add error handling for index deletion The index deletion during force resume lacks error handling. If the delete operation fails (e.g., due to concurrent operations or cluster state issues), the force resume process will fail without proper cleanup. Wrap the deletion in a try-catch block to handle failures gracefully and log appropriate error messages. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [173-178] if (clusterService.state().routingTable.hasIndex(followerIndexName)) { - client.suspending(client.admin().indices()::delete, defaultContext = true)( - org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) - ) - log.info("Force resume: deleted standalone follower index $followerIndexName") + try { + client.suspending(client.admin().indices()::delete, defaultContext = true)( + org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) + ) + log.info("Force resume: deleted standalone follower index $followerIndexName") + } catch (e: Exception) { + log.error("Force resume: failed to delete follower index $followerIndexName", e) + throw IllegalStateException("Failed to delete follower index during force resume: ${e.message}", e) + } } Suggestion importance[1-10]: 7 __ Why: Adding error handling for the index deletion operation improves robustness by catching potential failures during force resume. However, the suggestion wraps the operation in a try-catch that may not be necessary since the calling context already has error handling via `listener.completeWith`, which will catch and propagate exceptions appropriately.	Medium
	Add error handling for stop operation The stop replication call during force resume lacks error handling. If the stop operation fails, the subsequent delete and start operations will execute on an index that may still have active replication tasks, leading to inconsistent state. Add error handling to ensure the stop operation completes successfully before proceeding. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [167-168] -client.suspendExecute(STOP_REPLICATION_ACTION_TYPE, StopIndexReplicationRequest(followerIndexName), defaultContext = true) -log.info("Force resume: replication stopped for $followerIndexName") +try { + client.suspendExecute(STOP_REPLICATION_ACTION_TYPE, StopIndexReplicationRequest(followerIndexName), defaultContext = true) + log.info("Force resume: replication stopped for $followerIndexName") +} catch (e: Exception) { + log.error("Force resume: failed to stop replication for $followerIndexName", e) + throw IllegalStateException("Failed to stop replication during force resume: ${e.message}", e) +} Suggestion importance[1-10]: 7 __ Why: Adding explicit error handling for the stop operation could improve error messages and logging. However, the `executeForceResume` function is already called within a `listener.completeWith` block (line 83) that handles exceptions, so the additional try-catch may be redundant. The suggestion is valid but offers marginal improvement.	Medium
	Add error handling for start operation The start replication call during force resume lacks error handling. If the start operation fails after the index has been deleted, the follower index will be left in a broken state with no replication and no index. Add error handling to catch failures and provide clear error messages to help with recovery. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [193-194] -client.suspendExecute(ReplicateIndexAction.INSTANCE, startRequest, defaultContext = true) -log.info("Force resume: replication restarted for $followerIndexName") +try { + client.suspendExecute(ReplicateIndexAction.INSTANCE, startRequest, defaultContext = true) + log.info("Force resume: replication restarted for $followerIndexName") +} catch (e: Exception) { + log.error("Force resume: failed to restart replication for $followerIndexName", e) + throw IllegalStateException("Failed to restart replication during force resume. Index may need manual recovery: ${e.message}", e) +} Suggestion importance[1-10]: 7 __ Why: Adding error handling for the start replication operation could provide better error messages when the operation fails. However, like the previous suggestions, this is already within a `listener.completeWith` block that handles exceptions. The suggestion is valid but provides only marginal improvement over the existing error handling mechanism.	Medium

Suggestions up to commit b7f4ab9

Category	Suggestion	Impact
Possible issue	Verify replication started successfully The force resume path doesn't verify that the start replication operation completed successfully. If the start fails (e.g., snapshot restore issues, permission problems), the user receives an acknowledgment but replication isn't actually running. Add validation after start to ensure replication reached an active state before returning success. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [193-194] client.suspendExecute(ReplicateIndexAction.INSTANCE, startRequest, defaultContext = true) log.info("Force resume: replication restarted for $followerIndexName") +// Verify replication actually started +assertBusy { + val state = replicationMetadataManager.getIndexReplicationMetadata(followerIndexName) + require(state.overallState == ReplicationOverallState.RUNNING) { + "Replication failed to start after force resume" + } +} + Suggestion importance[1-10]: 6 __ Why: Verifying that replication reached `RUNNING` state before returning success improves reliability. However, the `improved_code` uses `assertBusy` without proper imports or timeout handling, and the verification logic may need adjustment for the asynchronous nature of replication startup.	Low
	Handle stop replication failures The stop replication call in force resume doesn't handle potential failures. If stop fails (e.g., due to concurrent operations or cluster state issues), the subsequent delete and start operations will proceed anyway, potentially leaving the system in an inconsistent state. Wrap the stop call in error handling to abort force resume on failure. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [167-168] -client.suspendExecute(STOP_REPLICATION_ACTION_TYPE, StopIndexReplicationRequest(followerIndexName), defaultContext = true) -log.info("Force resume: replication stopped for $followerIndexName") +try { + client.suspendExecute(STOP_REPLICATION_ACTION_TYPE, StopIndexReplicationRequest(followerIndexName), defaultContext = true) + log.info("Force resume: replication stopped for $followerIndexName") +} catch (e: Exception) { + log.error("Force resume failed during stop for $followerIndexName", e) + throw IllegalStateException("Failed to stop replication during force resume: ${e.message}", e) +} Suggestion importance[1-10]: 5 __ Why: Adding error handling for the `stop` operation improves robustness by preventing subsequent operations if stop fails. However, the `suspendExecute` call already propagates exceptions, so the try-catch mainly adds explicit logging and a custom error message.	Low
	Verify index state before deletion The force resume path deletes the follower index without verifying it's in a safe state. If the index is actively being written to or has ongoing operations, deletion could fail or cause data inconsistency. Add a check to ensure the index is closed or has no active operations before deletion. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [173-178] if (clusterService.state().routingTable.hasIndex(followerIndexName)) { + val indexMetadata = clusterService.state().metadata.index(followerIndexName) + if (indexMetadata?.state != IndexMetadata.State.CLOSE) { + client.suspending(client.admin().indices()::close, defaultContext = true)( + org.opensearch.action.admin.indices.close.CloseIndexRequest(followerIndexName) + ) + } client.suspending(client.admin().indices()::delete, defaultContext = true)( org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) ) log.info("Force resume: deleted standalone follower index $followerIndexName") } Suggestion importance[1-10]: 3 __ Why: The suggestion to close the index before deletion adds safety, but after `stop` replication, the follower index is standalone with no active replication tasks. The deletion should be safe without explicit closure. The added complexity may not be necessary.	Low

Suggestions up to commit 6da406c

Category	Suggestion	Impact
Possible issue	Verify force resume completion The force resume path returns immediately after `executeForceResume` without verifying that replication actually reached a healthy state. If the internal stop or start fails silently, the user receives a success response despite replication not being resumed. Add error handling to `executeForceResume` and verify the final replication state before returning success. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [102-108] if (!resumable && request.forceResume) { // Force resume via internal stop + start. // This reuses the battle-tested stop and start paths, ensuring proper // cleanup of metadata, checkpoints, tasks, and index blocks. - executeForceResume(request.indexName, replMetadata, remoteMetadata) + try { + executeForceResume(request.indexName, replMetadata, remoteMetadata) + // Verify replication state after force resume + val finalState = replicationMetadataManager.getIndexReplicationMetadata(request.indexName) + if (finalState.state != ReplicationOverallState.RUNNING) { + throw IllegalStateException("Force resume completed but replication is not in RUNNING state") + } + } catch (e: Exception) { + log.error("Force resume failed for ${request.indexName}", e) + throw e + } return@completeWith AcknowledgedResponse(true) } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that `executeForceResume` returns immediately without verifying the final state. However, the force resume path involves asynchronous operations (stop, delete, start), and the suggested synchronous state check may not be reliable immediately after `executeForceResume` completes. The concern is valid but the proposed solution may need refinement for async contexts.	Medium
Possible issue	Handle follower index deletion failure Deleting the follower index during force resume can cause data loss if the subsequent start replication fails (network issues, permission errors, etc.). The user loses both the old follower data and fails to establish new replication. Consider creating a snapshot of the follower index before deletion or implementing a rollback mechanism to restore the previous state on failure. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [170-178] // Step 2: Delete the follower index so start replication can create it fresh. // After stop, the follower is a standalone index. Start replication requires // the follower index to NOT exist (it creates it via snapshot restore). if (clusterService.state().routingTable.hasIndex(followerIndexName)) { - client.suspending(client.admin().indices()::delete, defaultContext = true)( - org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) - ) - log.info("Force resume: deleted standalone follower index $followerIndexName") + try { + client.suspending(client.admin().indices()::delete, defaultContext = true)( + org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) + ) + log.info("Force resume: deleted standalone follower index $followerIndexName") + } catch (e: Exception) { + log.error("Force resume: failed to delete follower index $followerIndexName", e) + throw IllegalStateException("Cannot proceed with force resume: failed to delete existing follower index", e) + } } Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid concern about data loss if deletion succeeds but subsequent start fails. However, the `suspending` call already propagates exceptions, so explicit try-catch adds limited value. The broader concern about rollback mechanisms is valid but beyond the scope of simple error handling.	Low

Suggestions up to commit 8e877db

Category	Suggestion	Impact
Possible issue	Handle lease acquisition failures properly The pre-restore lease acquisition loop catches only `RetentionLeaseAlreadyExistsException` but ignores all other potential failures. If lease acquisition fails for any shard (network issues, leader shard unavailable, etc.), the loop continues silently, leaving some shards without leases. This creates a partial lease state that will cause replication to fail later. Fail fast on any lease acquisition error. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [227-240] for (shardId in 0 until leaderIndexMetadata.numberOfShards) { val leaderShardId = ShardId(leaderIndexMetadata.index, shardId) val retainingSeqNo = globalCheckpoints.getValue(shardId) + 1 val syntheticFollowerShardId = ShardId( org.opensearch.core.index.Index(followerIndexName, "_na_"), shardId ) try { retentionLeaseHelper.addRetentionLeaseAsync(leaderShardId, retainingSeqNo, syntheticFollowerShardId) } catch (e: RetentionLeaseAlreadyExistsException) { retentionLeaseHelper.renewRetentionLease(leaderShardId, retainingSeqNo, syntheticFollowerShardId) + } catch (e: Exception) { + log.error("Force resume: failed to acquire lease for shard $shardId", e) + throw IllegalStateException("Cannot proceed with force resume: failed to acquire retention lease for shard $shardId", e) } log.info("Pre-restore lease acquired for shard $shardId at seqNo $retainingSeqNo") } Suggestion importance[1-10]: 8 __ Why: This is a critical issue. The code only catches `RetentionLeaseAlreadyExistsException` but silently continues on other failures, which could leave shards without leases and cause replication to fail later. The suggestion to fail fast on any lease acquisition error is correct and important for preventing partial lease states.	Medium
	Add error handling for index deletion The follower index deletion during force resume lacks proper error handling. If the delete operation fails (e.g., due to concurrent operations or cluster state issues), the force resume will proceed to acquire leases and start replication, potentially causing data inconsistency or replication failure. Add explicit error handling and verification that the index is fully deleted before proceeding. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [176-181] if (clusterService.state().routingTable.hasIndex(followerIndexName)) { - client.suspending(client.admin().indices()::delete, defaultContext = true)( - org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) - ) - log.info("Force resume: deleted standalone follower index $followerIndexName") + try { + client.suspending(client.admin().indices()::delete, defaultContext = true)( + org.opensearch.action.admin.indices.delete.DeleteIndexRequest(followerIndexName) + ) + // Verify deletion completed + assertBusy { + if (clusterService.state().routingTable.hasIndex(followerIndexName)) { + throw IllegalStateException("Index $followerIndexName still exists after deletion") + } + } + log.info("Force resume: deleted standalone follower index $followerIndexName") + } catch (e: Exception) { + log.error("Force resume: failed to delete follower index $followerIndexName", e) + throw IllegalStateException("Cannot proceed with force resume: failed to delete follower index $followerIndexName", e) + } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that the index deletion lacks error handling. However, the `improved_code` introduces `assertBusy` which is a test utility and shouldn't be used in production code. The core concern about handling deletion failures is valid and important for data consistency.	Medium
	Prevent null pointer on seqNoStats access The global checkpoint extraction assumes `seqNoStats` is always non-null when a primary shard is found, but `seqNoStats` can be null if the shard is initializing or recovering. This will cause a `NullPointerException` when accessing `globalCheckpoint`. Add an explicit null check for `seqNoStats` before accessing its properties. src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt [253-262] val checkpoints = mutableMapOf<Int, Long>() for (shardId in 0 until numberOfShards) { - val checkpoint = statsResponse.shards + val shardStats = statsResponse.shards .firstOrNull { it.shardRouting.shardId().id == shardId && it.shardRouting.primary() } - ?.seqNoStats?.globalCheckpoint ?: throw IllegalStateException( - "Primary shard $shardId not found or seqNoStats unavailable for leader index $leaderIndexName" + "Primary shard $shardId not found for leader index $leaderIndexName" + ) + val checkpoint = shardStats.seqNoStats?.globalCheckpoint + ?: throw IllegalStateException( + "seqNoStats unavailable for primary shard $shardId of leader index $leaderIndexName" ) checkpoints[shardId] = checkpoint } Suggestion importance[1-10]: 7 __ Why: The suggestion identifies a potential `NullPointerException` when `seqNoStats` is null during shard initialization or recovery. The improved code properly separates the null checks for the shard and `seqNoStats`, making the error messages more specific and preventing the NPE.	Medium

github-actions · 2026-05-10T15:02:29Z

Persistent review updated to latest commit 5017605

mohit10011999 · 2026-05-10T15:03:04Z

Regarding #1685 (comment)
and #1685 (comment) AI comment

Regarding Race Condition comment:-
No - retention leases prevent purging by design. Once acquired at seqNo X, operations ≥ X are retained regardless of global checkpoint advances.

Incomplete Cleanup
Added a log warning + continue instead of silent skip

Logic Error
No - the bot misread the condition. if (!resumable && request.forceResume) means the coordinator only runs when leases are missing. If leases exist, it's skipped.

Null seqNoStats
Improved error message

github-actions · 2026-05-10T15:57:40Z

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit 82ea354.

Path	Line	Severity	Description
src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt	183	medium	Force resume path sets `isAutoFollowRequest = true` on the internal ReplicateIndexRequest, explicitly bypassing setup validation checks (comment: 'Skip setup checks since this is an internal re-start'). If the force_resume API endpoint has weaker authorization than the normal start-replication endpoint, this could allow a caller to trigger a replication start that skips security/permission validation they would otherwise fail. The justification given is plausible for the feature, but the bypass is unconditional and warrants verifying that the force_resume API enforces equivalent authorization to the original startReplication call.
src/main/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt	170	low	The force resume path unconditionally deletes the follower index (a permanent destructive action) as a side effect, with only an internal log message and no explicit user-visible warning surfaced in the API response. A caller sending `force_resume=true` may not realize the follower index and all its data will be wiped and rebuilt from snapshot. This is a design transparency issue rather than evidence of malicious intent, but data loss without a confirmation step warrants review.

The table above displays the top 10 most important findings.

Total: 2 | Critical: 0 | High: 0 | Medium: 1 | Low: 1

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

github-actions · 2026-05-10T15:58:32Z

Persistent review updated to latest commit a23f7a8

mohit10011999 · 2026-05-11T05:12:30Z

#1685 (comment)
Addressed as below:-

Replaced .all() with .clear() followed by .translog(true) — this tells the leader to only compute translog-related stats, dramatically reducing the payload sent across the cluster boundary.
seqNoStats is always available at the shard level regardless of flags, so the globalCheckpoint read still works correctly.

github-actions · 2026-05-11T05:13:56Z

Persistent review updated to latest commit fdf3669

soosinha · 2026-05-27T08:19:45Z

Can you add ITs ?

github-actions · 2026-05-27T09:59:12Z

Persistent review updated to latest commit a22e956

github-actions · 2026-05-28T04:04:01Z

Persistent review updated to latest commit fc1a080

github-actions · 2026-05-28T04:19:19Z

Persistent review updated to latest commit 0f53dec

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

- Removed reimplemented block removal/addition (reuses UpdateIndexBlockAction, same as stop action) - Removed reimplemented index deletion (reuses same pattern as IndexReplicationTask.cancelRestore) - Removed reimplemented lease cleanup (reuses existing RemoteClusterRetentionLeaseHelper methods) - Removed ReplicationMetadataManager dependency from coordinator (not needed) - Simplified getLeaderGlobalCheckpoint using firstOrNull (same API as RemoteClusterRepository) - Only genuinely new logic retained: pre-restore lease acquisition at leaderCheckpoint+1 Signed-off-by: Mohit Kumar <mohitamg@amazon.com>

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

github-actions · 2026-05-30T13:20:07Z

Persistent review updated to latest commit d63504b

github-actions · 2026-05-30T13:30:30Z

Persistent review updated to latest commit a1901ff

github-actions · 2026-05-30T13:32:21Z

Persistent review updated to latest commit d9fc56e

github-actions · 2026-05-30T14:15:06Z

Persistent review updated to latest commit 28878c9

github-actions · 2026-05-30T14:39:38Z

Persistent review updated to latest commit 88aa4ba

github-actions · 2026-05-30T14:55:13Z

Persistent review updated to latest commit 8e877db

github-actions · 2026-05-30T15:26:07Z

Persistent review updated to latest commit 6da406c

github-actions · 2026-05-30T15:44:15Z

Persistent review updated to latest commit b7f4ab9

github-actions · 2026-05-30T16:21:40Z

Persistent review updated to latest commit 576641a

github-actions · 2026-05-30T17:01:43Z

Persistent review updated to latest commit 2bf84b4

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

github-actions · 2026-06-01T10:49:12Z

Persistent review updated to latest commit 82ea354

mohit10011999 · 2026-06-09T04:18:37Z

Test Coverage for Force Resume

Unit Tests — `ResumeIndexReplicationRequestTests.kt`

#	Scenario	What it validates
1	Serialization roundtrip with forceResume=false	Wire format backward compatibility — default false survives serialize/deserialize
2	Serialization roundtrip with forceResume=true	Wire format carries force_resume=true correctly across nodes
3	indices() returns index name	Request correctly identifies target index
4	validate() returns null	No validation errors for valid request
5	fromXContent with force_resume=true	REST body `{"force_resume": true}` parsed correctly
6	fromXContent with force_resume=false	REST body `{"force_resume": false}` parsed correctly
7	fromXContent with empty body defaults to false	Empty `{}` body maintains backward compatibility
8	toXContent includes force_resume field	API response includes the field
9	indicesOptions is strict single index	Request targets exactly one index

Unit Tests — `TransportResumeIndexReplicationActionTests.kt`

#	Scenario	What it validates
1	Leases valid + forceResume=true → skips force path	No unnecessary stop+start when leases are healthy
2	Leases expired + forceResume=false → throws	User gets clear error instead of silent failure
3	Leases expired + forceResume=true → triggers force path	Core routing logic works correctly
4	Role preservation with both roles present	Follower/leader roles carried into start request
5	Role preservation skipped when roles are null	No NPE when roles aren't set

Integration Tests — `ForceResumeReplicationIT.kt`

#	Scenario	What it validates
1	Force resume after retention lease expires	Full flow: start → sync → pause → expire leases (1s period, 30s wait) → normal resume fails → force resume succeeds → reaches SYNCING/BOOTSTRAPPING state
2	Force resume when retention leases still exist	force_resume=true with valid leases falls through to normal resume path — no unnecessary stop+start
3	Error message suggests force_resume option	When normal resume fails due to expired leases, error contains "force_resume=true" hint
4	Data written after force resume is replicated	5 docs before pause + 5 docs during pause → force resume → all 10 present on follower → 5 more post-resume replicate (15 total)
5	Force resume on multi-shard index (5 shards)	20 docs across 5 shards → pause → expire leases → 10 more during pause → force resume → all 30 present → 5 more post-resume replicate (35 total)

mohit10011999 requested review from ankitkala, gbbafna, krishna-ggk, monusingh-1, saikaranam-amazon and soosinha as code owners May 9, 2026 13:00

mohit10011999 force-pushed the ForceResume branch 2 times, most recently from c20e107 to fb17424 Compare May 9, 2026 13:07

mohit10011999 force-pushed the ForceResume branch from a8e76d6 to 5017605 Compare May 10, 2026 15:01

mohit10011999 force-pushed the ForceResume branch from 5017605 to a23f7a8 Compare May 10, 2026 15:56

soosinha reviewed May 27, 2026

View reviewed changes

Comment thread ...ain/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt Outdated

Comment thread src/main/kotlin/org/opensearch/replication/seqno/RemoteClusterRetentionLeaseHelper.kt Outdated

mohit10011999 force-pushed the ForceResume branch from e5aa8d4 to a22e956 Compare May 27, 2026 09:57

mohit10011999 force-pushed the ForceResume branch from fc1a080 to 0f53dec Compare May 28, 2026 04:17

mohit10011999 mentioned this pull request May 28, 2026

[DOC] Add documentation for force_resume parameter for Cross-Cluster Replication Resume API opensearch-project/documentation-website#12476

Closed

4 tasks

ankitkala reviewed May 30, 2026

View reviewed changes

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

mohitamg added 3 commits May 30, 2026 18:45

Add force resume coordinator and update resume replication flow

62923fb

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

Minor: clean up comment formatting in ForceResumeCoordinator

a518679

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

mohit10011999 force-pushed the ForceResume branch from 00fafb3 to d63504b Compare May 30, 2026 13:18

mohit10011999 commented May 30, 2026

View reviewed changes

Comment thread ...ain/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt

mohit10011999 force-pushed the ForceResume branch from d63504b to a1901ff Compare May 30, 2026 13:28

mohit10011999 force-pushed the ForceResume branch from a1901ff to d9fc56e Compare May 30, 2026 13:30

mohit10011999 force-pushed the ForceResume branch from d9fc56e to 28878c9 Compare May 30, 2026 14:13

mohit10011999 force-pushed the ForceResume branch from 28878c9 to 88aa4ba Compare May 30, 2026 14:37

mohit10011999 force-pushed the ForceResume branch from 88aa4ba to 8e877db Compare May 30, 2026 14:53

mohit10011999 force-pushed the ForceResume branch from 8e877db to 6da406c Compare May 30, 2026 15:24

mohit10011999 force-pushed the ForceResume branch from 6da406c to b7f4ab9 Compare May 30, 2026 15:42

mohit10011999 force-pushed the ForceResume branch from b7f4ab9 to 576641a Compare May 30, 2026 16:20

mohit10011999 force-pushed the ForceResume branch from 576641a to 2bf84b4 Compare May 30, 2026 17:00

Refactor force resume to use internal stop+start instead of custom co…

82ea354

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

mohit10011999 force-pushed the ForceResume branch from 2bf84b4 to 82ea354 Compare June 1, 2026 10:47

mohit10011999 mentioned this pull request Jun 1, 2026

Add force resume parameter documentation for cross-cluster replication resume API opensearch-project/documentation-website#12492

Merged

1 task

soosinha approved these changes Jun 9, 2026

View reviewed changes

mohit10011999 merged commit 2c6dbf6 into opensearch-project:main Jun 9, 2026
17 checks passed

mohit10011999 self-assigned this Jun 9, 2026

soosinha reviewed Jun 9, 2026

View reviewed changes

Comment thread ...ain/kotlin/org/opensearch/replication/action/resume/TransportResumeIndexReplicationAction.kt

Uh oh!

Conversation

mohit10011999 commented May 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Solution

Backward Compatibility

Related Issues

Check List

Uh oh!

github-actions Bot commented May 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Reviewer Guide 🔍

(Review updated until commit 82ea354)

Uh oh!

github-actions Bot commented May 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Code Suggestions ✨

Previous suggestions

Uh oh!

github-actions Bot commented May 10, 2026

Uh oh!

mohit10011999 commented May 10, 2026

Uh oh!

github-actions Bot commented May 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Code Analyzer ❗

Uh oh!

github-actions Bot commented May 10, 2026

Uh oh!

mohit10011999 commented May 11, 2026

Uh oh!

github-actions Bot commented May 11, 2026

Uh oh!

Uh oh!

Uh oh!

soosinha commented May 27, 2026

Uh oh!

github-actions Bot commented May 27, 2026

Uh oh!

github-actions Bot commented May 28, 2026

Uh oh!

github-actions Bot commented May 28, 2026

Uh oh!

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented May 30, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

mohit10011999 commented Jun 9, 2026

Test Coverage for Force Resume

Unit Tests — ResumeIndexReplicationRequestTests.kt

Unit Tests — TransportResumeIndexReplicationActionTests.kt

Integration Tests — ForceResumeReplicationIT.kt

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

mohit10011999 commented May 9, 2026 •

edited

Loading

github-actions Bot commented May 10, 2026 •

edited

Loading

(Review updated until commit `82ea354`)

github-actions Bot commented May 10, 2026 •

edited

Loading

github-actions Bot commented May 10, 2026 •

edited

Loading

Unit Tests — `ResumeIndexReplicationRequestTests.kt`

Unit Tests — `TransportResumeIndexReplicationActionTests.kt`

Integration Tests — `ForceResumeReplicationIT.kt`