Focused on Add a validation in Start Replication API to fail the call incase there is any stale replication metadata and Stop Replication API should ensure that it removes all stale replication metadata from the cluster state upon execution.

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

Author: mohitamg

src/main/kotlin/org/opensearch/replication/ReplicationEngine.kt

@@ -11,7 +11,9 @@
 
 package org.opensearch.replication
 
+import org.opensearch.index.engine.DeletionStrategy
 import org.opensearch.index.engine.EngineConfig
+import org.opensearch.index.engine.IndexingStrategy
 import org.opensearch.index.engine.InternalEngine
 import org.opensearch.index.seqno.SequenceNumbers
 

src/main/kotlin/org/opensearch/replication/action/index/TransportReplicateIndexAction.kt

@@ -15,8 +15,6 @@ import org.opensearch.replication.ReplicationPlugin
 import org.opensearch.replication.action.setup.SetupChecksAction
 import org.opensearch.replication.action.setup.SetupChecksRequest
 import org.opensearch.replication.metadata.store.ReplicationContext
-import org.opensearch.replication.task.cleanup.StaleArtifactDetector
-import org.opensearch.replication.task.cleanup.TaskCleanupManager
 import org.opensearch.replication.util.SecurityContext
 import org.opensearch.replication.util.ValidationUtil
 import org.opensearch.replication.util.completeWith
@@ -54,9 +52,7 @@ class TransportReplicateIndexAction @Inject constructor(transportService: Transp
                                                         actionFilters: ActionFilters,
                                                         private val client : Client,
                                                         private val environment: Environment,
-                                                        private val metadataCreateIndexService: MetadataCreateIndexService,
-                                                        private val staleArtifactDetector: StaleArtifactDetector,
-                                                        private val taskCleanupManager: TaskCleanupManager) :
+                                                        private val metadataCreateIndexService: MetadataCreateIndexService) :
         HandledTransportAction<ReplicateIndexRequest, ReplicateIndexResponse>(ReplicateIndexAction.NAME,
                 transportService, actionFilters, ::ReplicateIndexRequest),
     CoroutineScope by GlobalScope {
@@ -69,10 +65,8 @@ class TransportReplicateIndexAction @Inject constructor(transportService: Transp
         launch(threadPool.coroutineContext()) {
             listener.completeWith {
                 log.info("Setting-up replication for ${request.leaderAlias}:${request.leaderIndex} -> ${request.followerIndex}")
-                
-                performPreOperationValidation(request.followerIndex)
-                
                 val user = SecurityContext.fromSecurityThreadContext(threadPool.threadContext)
+
                 val followerReplContext = ReplicationContext(request.followerIndex,
                         user?.overrideFgacRole(request.useRoles?.get(ReplicateIndexRequest.FOLLOWER_CLUSTER_ROLE)))
                 val leaderReplContext = ReplicationContext(request.leaderIndex,
@@ -99,12 +93,15 @@ class TransportReplicateIndexAction @Inject constructor(transportService: Transp
 
                 val leaderSettings = getLeaderIndexSettings(request.leaderAlias, request.leaderIndex)
                 log.debug("Leader settings were fetched for ${request.leaderIndex} index.")
+
                 if (leaderSettings.keySet().contains(ReplicationPlugin.REPLICATED_INDEX_SETTING.key) and
                         !leaderSettings.get(ReplicationPlugin.REPLICATED_INDEX_SETTING.key).isNullOrBlank()) {
                     throw IllegalArgumentException("Cannot Replicate a Replicated Index ${request.leaderIndex}")
                 }
+
+                // Soft deletes should be enabled for replication to work.
                 if (!leaderSettings.getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.key, false)) {
-                    throw IllegalArgumentException("Cannot Replicate an index where ${IndexSettings.INDEX_SOFT_DELETES_SETTING.key} is disabled")
+                    throw IllegalArgumentException("Cannot Replicate an index where the setting ${IndexSettings.INDEX_SOFT_DELETES_SETTING.key} is disabled")
                 }
 
                 // Disabling knn checks as new api call will require us add roles in security index which will be a breaking call.
@@ -168,61 +165,4 @@ class TransportReplicateIndexAction @Inject constructor(transportService: Transp
         // and then the explicitly set ones to override the default settings.
         return Settings.builder().put(leaderDefaultSettings).put(leaderSettings).build()
     }
-
-    /**
-     * This method distinguishes between active replication (RUNNING/PAUSED state)
-     * and truly stale artifacts from failed operations. Active replication will cause this
-     * method to throw an error instead of attempting cleanup.
-     */
-    private suspend fun performPreOperationValidation(followerIndex: String) {
-        // If the follower index already exists, skip this validation
-        // The index existence check in TransportReplicateIndexClusterManagerNodeAction will handle it
-        if (clusterService.state().metadata().hasIndex(followerIndex)) {
-            return
-        }
-        
-        // First check if replication is already active for this index
-        val replicationStateParams = org.opensearch.replication.metadata.state.getReplicationStateParamsForIndex(
-            clusterService, 
-            followerIndex
-        )
-        
-        if (replicationStateParams != null) {
-            val currentState = replicationStateParams[org.opensearch.replication.metadata.state.REPLICATION_LAST_KNOWN_OVERALL_STATE]
-            
-            // If replication is RUNNING or PAUSED, it's active - don't cleanup, throw error
-            if (currentState == org.opensearch.replication.metadata.ReplicationOverallState.RUNNING.name || 
-                currentState == org.opensearch.replication.metadata.ReplicationOverallState.PAUSED.name) {
-                throw IllegalStateException(
-                    "Replication is already active for index $followerIndex in $currentState state. " +
-                    "Cannot start replication again. Use resume API to restart paused replication."
-                )
-            }
-            
-            // If state is STOPPED or FAILED, artifacts are stale and should be cleaned up
-            log.info("Found replication metadata for $followerIndex in $currentState state - will cleanup stale artifacts")
-        }
-        
-        // Now safe to detect and cleanup truly stale artifacts
-        val staleArtifactReport = staleArtifactDetector.detectStaleArtifacts(followerIndex)
-        
-        if (!staleArtifactReport.hasStaleArtifacts) return
-        
-        log.warn("Detected ${staleArtifactReport.artifacts.size} stale artifacts for $followerIndex")
-        log.info("Attempting cleanup of stale artifacts")
-        
-        val cleanupResult = taskCleanupManager.cleanupAllReplicationTasks(followerIndex)
-        
-        if (!cleanupResult.success) {
-            val failures = cleanupResult.failures.joinToString("; ") { "${it.component}: ${it.error}" }
-            throw IllegalStateException(
-                "Cannot start replication for $followerIndex due to conflicting stale tasks. " +
-                "Cleanup failed: $failures. Please manually resolve these conflicts."
-            )
-        }
-        
-        log.info("Cleaned up stale artifacts: ${cleanupResult.indexTasksRemoved} index tasks, " +
-                "${cleanupResult.shardTasksRemoved} shard tasks, ${cleanupResult.retentionLeasesRemoved} leases, " +
-                "${cleanupResult.persistentTasksRemoved} persistent tasks")
-    }
 }

src/main/kotlin/org/opensearch/replication/action/index/TransportReplicateIndexClusterManagerNodeAction.kt

@@ -45,6 +45,9 @@ import org.opensearch.common.settings.IndexScopedSettings
 import org.opensearch.index.IndexNotFoundException
 import org.opensearch.persistent.PersistentTasksService
 import org.opensearch.replication.ReplicationPlugin
+import org.opensearch.replication.metadata.state.REPLICATION_LAST_KNOWN_OVERALL_STATE
+import org.opensearch.replication.metadata.state.getReplicationStateParamsForIndex
+import org.opensearch.replication.task.cleanup.InlineStaleTaskCleaner
 import org.opensearch.replication.util.stackTraceToString
 import org.opensearch.repositories.RepositoriesService
 import org.opensearch.core.rest.RestStatus
@@ -91,6 +94,36 @@ class TransportReplicateIndexClusterManagerNodeAction @Inject constructor(transp
         // for each shard. If that takes too long we can start the task asynchronously and return the response first.
         launch(Dispatchers.Unconfined + threadPool.coroutineContext()) {
             try {
+                // Validate no active replication metadata exists for this index
+                val replicationStateParams = getReplicationStateParamsForIndex(clusterService, replicateIndexReq.followerIndex)
+                if (replicationStateParams != null) {
+                    val currentState = replicationStateParams[REPLICATION_LAST_KNOWN_OVERALL_STATE]
+                    if (currentState == ReplicationOverallState.RUNNING.name ||
+                        currentState == ReplicationOverallState.PAUSED.name) {
+                        throw IllegalStateException(
+                            "Replication is already active for index ${replicateIndexReq.followerIndex} in $currentState state. " +
+                            "Cannot start replication again. Use resume API to restart paused replication."
+                        )
+                    }
+                    // STOPPED or FAILED state indicates stale metadata
+                    throw IllegalStateException(
+                        "Stale replication metadata exists for index ${replicateIndexReq.followerIndex} in $currentState state. " +
+                        "Please run the Stop Replication API to clean up before starting replication."
+                    )
+                }
+
+                // Clean up any stale persistent tasks before creating new ones
+                try {
+                    val cleaner = InlineStaleTaskCleaner(clusterService, nodeClient)
+                    val removed = cleaner.removeStaleTasksForIndex(replicateIndexReq.followerIndex)
+                    if (removed > 0) {
+                        log.info("Cleaned up $removed stale task(s) for index ${replicateIndexReq.followerIndex} before starting replication")
+                    }
+                } catch (e: Exception) {
+                    log.error("Failed to clean up stale tasks for index ${replicateIndexReq.followerIndex}: ${e.message}", e)
+                    throw e
+                }
+
                 if(clusterService.clusterSettings.get(ReplicationPlugin.REPLICATION_FOLLOWER_BLOCK_START)) {
                     log.debug("Replication cannot be started as " +
                             "start block(${ReplicationPlugin.REPLICATION_FOLLOWER_BLOCK_START}) is set")