# This is a combination of 4 commits.

# This is the 1st commit message:

Removed unnecessary user notifications for version conflict exceptions in Snapshot Management

Signed-off-by: Kamal Nayan <askkamal@amazon.com>

# This is the commit message #2:

Unwrapped the exception

Signed-off-by: Kamal Nayan <askkamal@amazon.com>

# This is the commit message #3:

Added the unit tests and updated the code

Signed-off-by: Kamal Nayan <askkamal@amazon.com>

# This is the commit message #4:

Updated the code for minor fixes and nits

Signed-off-by: Kamal Nayan <askkamal@amazon.com>

Author: Kamal Nayan

src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/engine/SMStateMachine.kt

@@ -7,10 +7,12 @@ package org.opensearch.indexmanagement.snapshotmanagement.engine
 
 import org.apache.logging.log4j.LogManager
 import org.apache.logging.log4j.Logger
+import org.opensearch.ExceptionsHelper
 import org.opensearch.action.bulk.BackoffPolicy
 import org.opensearch.common.settings.Settings
 import org.opensearch.common.unit.TimeValue
 import org.opensearch.commons.ConfigConstants
+import org.opensearch.index.engine.VersionConflictEngineException
 import org.opensearch.indexmanagement.IndexManagementIndices
 import org.opensearch.indexmanagement.opensearchapi.IndexManagementSecurityContext
 import org.opensearch.indexmanagement.opensearchapi.retry
@@ -215,6 +217,14 @@ class SMStateMachine(
                 metadata = md
             }
         } catch (ex: Exception) {
+            val unwrappedException = ExceptionsHelper.unwrapCause(ex) as Exception
+            if (unwrappedException is VersionConflictEngineException) {
+                // Don't throw the exception
+                // TODO: Extract seqNo on VersionConflictException and retry updateMetadata with updated seqNo.
+                log.error("Version conflict exception while updating metadata.", ex)
+                return
+            }
+
             val smEx = SnapshotManagementException(ExceptionKey.METADATA_INDEXING_FAILURE, ex)
             log.error(smEx.message, ex)
             throw smEx
@@ -223,7 +233,7 @@ class SMStateMachine(
         // TODO SM save a copy to history
     }
 
-    private val updateMetaDataRetryPolicy =
+    val updateMetaDataRetryPolicy =
         BackoffPolicy.exponentialBackoff(
             TimeValue.timeValueMillis(EXPONENTIAL_BACKOFF_MILLIS), MAX_NUMBER_OF_RETRIES,
         )

src/test/kotlin/org/opensearch/indexmanagement/snapshotmanagement/engine/SMStateMachineTests.kt

@@ -6,12 +6,23 @@
 package org.opensearch.indexmanagement.snapshotmanagement.engine
 
 import com.nhaarman.mockitokotlin2.argumentCaptor
+import com.nhaarman.mockitokotlin2.doAnswer
+import com.nhaarman.mockitokotlin2.doReturn
+import com.nhaarman.mockitokotlin2.mock
 import com.nhaarman.mockitokotlin2.spy
 import com.nhaarman.mockitokotlin2.times
 import com.nhaarman.mockitokotlin2.verify
 import kotlinx.coroutines.runBlocking
+import org.apache.logging.log4j.LogManager
+import org.apache.logging.log4j.Logger
+import org.opensearch.OpenSearchException
+import org.opensearch.action.bulk.BackoffPolicy
 import org.opensearch.common.unit.TimeValue
+import org.opensearch.core.index.shard.ShardId
+import org.opensearch.index.engine.VersionConflictEngineException
 import org.opensearch.indexmanagement.MocksTestCase
+import org.opensearch.indexmanagement.opensearchapi.retry
+import org.opensearch.indexmanagement.snapshotmanagement.SnapshotManagementException
 import org.opensearch.indexmanagement.snapshotmanagement.engine.states.SMState
 import org.opensearch.indexmanagement.snapshotmanagement.engine.states.creationTransitions
 import org.opensearch.indexmanagement.snapshotmanagement.engine.states.deletionTransitions
@@ -230,4 +241,70 @@ open class SMStateMachineTests : MocksTestCase() {
                 assertEquals(1, firstValue.policyPrimaryTerm)
             }
         }
+
+    fun `test updateMetadata handles VersionConflictEngineException gracefully`() = runBlocking {
+        val initialMetadata = randomSMMetadata(
+            policySeqNo = 0,
+            policyPrimaryTerm = 0,
+        )
+        val smPolicy = randomSMPolicy(
+            seqNo = 1,
+            primaryTerm = 1,
+        )
+        val updatedMetadata = randomSMMetadata(
+            policySeqNo = 1,
+            policyPrimaryTerm = 1,
+        )
+
+        val mockBackoffPolicy = mock<BackoffPolicy>()
+        val stateMachineSpy = spy(SMStateMachine(client, smPolicy, initialMetadata, settings, threadPool, indicesManager))
+        doReturn(mockBackoffPolicy).`when`(stateMachineSpy).updateMetaDataRetryPolicy
+        val logger: Logger = LogManager.getLogger(javaClass)
+
+        doAnswer { throw VersionConflictEngineException(ShardId("index", "_na_", 1), "test", "message") }
+            .`when`(mockBackoffPolicy)
+            .retry(logger) { true }
+
+        // Verify VersionConflictEngineException is handled gracefully
+        try {
+            stateMachineSpy.updateMetadata(updatedMetadata)
+        } catch (e: Exception) {
+            fail("VersionConflictEngineException should be handled without throwing: ${e.message}")
+        }
+    }
+
+    fun `test updateMetadata throws SnapshotManagementException for other exceptions`() = runBlocking {
+        val initialMetadata = randomSMMetadata(
+            policySeqNo = 0,
+            policyPrimaryTerm = 0,
+        )
+        val smPolicy = randomSMPolicy(
+            seqNo = 1,
+            primaryTerm = 1,
+        )
+        val updatedMetadata = randomSMMetadata(
+            policySeqNo = 1,
+            policyPrimaryTerm = 1,
+        )
+
+        val mockBackoffPolicy = mock<BackoffPolicy>()
+        val stateMachineSpy = spy(SMStateMachine(client, smPolicy, initialMetadata, settings, threadPool, indicesManager))
+        doReturn(mockBackoffPolicy).`when`(stateMachineSpy).updateMetaDataRetryPolicy
+        val logger: Logger = LogManager.getLogger(javaClass)
+
+        val openSearchException = OpenSearchException("Test exception")
+        doAnswer { throw openSearchException }
+            .`when`(mockBackoffPolicy)
+            .retry(logger) { true }
+
+        // Verify OpenSearchException is wrapped in SnapshotManagementException
+        val thrownException = assertThrows(SnapshotManagementException::class.java) {
+            runBlocking {
+                stateMachineSpy.updateMetadata(updatedMetadata)
+            }
+        }
+
+        // Verify exception type and cause
+        assertTrue(thrownException.cause is OpenSearchException)
+    }
 }