Changes for stop-replication action

Signed-off-by: aggarwalShivani <shivani.aggarwal@nokia.com>

Author: aggarwalShivani

src/main/kotlin/org/opensearch/commons/replication/ReplicationPluginInterface.kt

@@ -5,8 +5,9 @@
 package org.opensearch.commons.replication
 
 import org.opensearch.action.support.master.AcknowledgedResponse
+import org.opensearch.client.Client
 import org.opensearch.client.node.NodeClient
-import org.opensearch.commons.replication.action.ReplicationActions.UNFOLLOW_REPLICATION_ACTION_TYPE
+import org.opensearch.commons.replication.action.ReplicationActions.INTERNAL_STOP_REPLICATION_ACTION_TYPE
 import org.opensearch.commons.replication.action.StopIndexReplicationRequest
 import org.opensearch.commons.utils.recreateObject
 import org.opensearch.core.action.ActionListener
@@ -16,7 +17,7 @@ import org.opensearch.core.common.io.stream.Writeable
 /**
  * Transport action plugin interfaces for the cross-cluster-replication plugin.
  */
-object ReplicationPluginInterface {
+open class ReplicationPluginInterface {
 
     /**
      * Stop replication.
@@ -25,13 +26,14 @@ object ReplicationPluginInterface {
      * @param listener The listener for getting response
      */
 
-    fun stopReplication(
-        client: NodeClient,
+    open fun stopReplication(
+        client: Client,
         request: StopIndexReplicationRequest,
         listener: ActionListener<AcknowledgedResponse>
     ) {
-        return client.execute(
-            UNFOLLOW_REPLICATION_ACTION_TYPE,
+        val nodeClient = client as NodeClient
+        return nodeClient.execute(
+            INTERNAL_STOP_REPLICATION_ACTION_TYPE,
             request,
             wrapActionListener(listener) { response ->
                 recreateObject(response) {

src/main/kotlin/org/opensearch/commons/replication/action/ReplicationActions.kt

@@ -14,18 +14,17 @@ object ReplicationActions {
 
     /**
      * Action names for stopping replication
-     * STOP_REPLICATION_ACTION_NAME: action used for _stop REST API
-     * UNFOLLOW_REPLICATION_ACTION_NAME: internal action used for inter-plugin communication i.e. by ism to invoke stop
-     * replication.
+     * STOP_REPLICATION_ACTION_NAME: action used for _replication/_stop REST API
+     * INTERNAL_STOP_REPLICATION_ACTION_NAME: Internal only - Used by Index Management plugin to invoke stop replication
      */
     const val STOP_REPLICATION_ACTION_NAME = "indices:admin/plugins/replication/index/stop"
-    const val UNFOLLOW_REPLICATION_ACTION_NAME = "indices:admin/plugins/replication/index/unfollow"
+    const val INTERNAL_STOP_REPLICATION_ACTION_NAME = "indices:internal/plugins/replication/index/stop"
 
     /**
      * Stop replication transport action types.
      */
     val STOP_REPLICATION_ACTION_TYPE =
         ActionType(STOP_REPLICATION_ACTION_NAME, ::AcknowledgedResponse)
-    val UNFOLLOW_REPLICATION_ACTION_TYPE =
-        ActionType(UNFOLLOW_REPLICATION_ACTION_NAME, ::AcknowledgedResponse)
+    val INTERNAL_STOP_REPLICATION_ACTION_TYPE =
+        ActionType(INTERNAL_STOP_REPLICATION_ACTION_NAME, ::AcknowledgedResponse)
 }

src/main/kotlin/org/opensearch/commons/replication/action/StopIndexReplicationRequest.kt

@@ -1,3 +1,7 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
 package org.opensearch.commons.replication.action
 
 import org.opensearch.action.ActionRequestValidationException

src/test/kotlin/org/opensearch/commons/replication/ReplicationPluginInterfaceTests.kt

@@ -8,37 +8,60 @@ package org.opensearch.commons.replication
 import com.nhaarman.mockitokotlin2.whenever
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.extension.ExtendWith
-import org.mockito.Answers
-import org.mockito.ArgumentMatchers
-import org.mockito.Mock
-import org.mockito.Mockito
+import org.mockito.Mockito.any
+import org.mockito.Mockito.verify
+import org.mockito.Mockito.mock
 import org.mockito.junit.jupiter.MockitoExtension
-import org.opensearch.action.ActionType
 import org.opensearch.action.support.master.AcknowledgedResponse
 import org.opensearch.client.node.NodeClient
 import org.opensearch.commons.replication.action.StopIndexReplicationRequest
 import org.opensearch.core.action.ActionListener
+import org.opensearch.core.action.ActionResponse
 
-@Suppress("UNCHECKED_CAST")
 @ExtendWith(MockitoExtension::class)
 internal class ReplicationPluginInterfaceTests {
 
-    @Mock(answer = Answers.RETURNS_DEEP_STUBS)
-    private lateinit var client: NodeClient
+    @Test
+    fun `test stopReplication successful response`() {
+        // Mock dependencies
+        val client: NodeClient = mock()
+        val request: StopIndexReplicationRequest = mock()
+        val listener: ActionListener<AcknowledgedResponse> = mock()
+        val acknowledgedResponse = AcknowledgedResponse(true) // Successful response
+
+        // Mock the behavior of NodeClient.execute()
+        whenever(client.execute(any(), any(), any<ActionListener<ActionResponse>>()))
+            .thenAnswer { invocation ->
+                val actionListener = invocation.getArgument<ActionListener<ActionResponse>>(2)
+                actionListener.onResponse(acknowledgedResponse) // Simulate success
+            }
+
+        val replicationPluginInterface = ReplicationPluginInterface()
+        // Call method under test
+        replicationPluginInterface.stopReplication(client, request, listener)
+        // Verify that listener.onResponse is called with the correct response
+        verify(listener).onResponse(acknowledgedResponse)
+    }
 
     @Test
-    fun stopReplication() {
-        val request = Mockito.mock(StopIndexReplicationRequest::class.java)
-        val response = AcknowledgedResponse(true)
-        val listener: ActionListener<AcknowledgedResponse> =
-            Mockito.mock(ActionListener::class.java) as ActionListener<AcknowledgedResponse>
-
-        Mockito.doAnswer {
-            (it.getArgument(2) as ActionListener<AcknowledgedResponse>)
-                .onResponse(response)
-        }.whenever(client).execute(Mockito.any(ActionType::class.java), Mockito.any(), Mockito.any())
-
-        ReplicationPluginInterface.stopReplication(client, request, listener)
-        Mockito.verify(listener, Mockito.times(1)).onResponse(ArgumentMatchers.eq(response))
+    fun `test stopReplication failure response`() {
+        // Mock dependencies
+        val client: NodeClient = mock()
+        val request: StopIndexReplicationRequest = mock()
+        val listener: ActionListener<AcknowledgedResponse> = mock()
+        val exception = Exception("Test failure")
+
+        // Mock the behavior of NodeClient.execute()
+        whenever(client.execute(any(), any(), any<ActionListener<ActionResponse>>()))
+            .thenAnswer { invocation ->
+                val actionListener = invocation.getArgument<ActionListener<ActionResponse>>(2)
+                actionListener.onFailure(exception) // Simulate failure
+            }
+
+        val replicationPluginInterface = ReplicationPluginInterface()
+        // Call method under test
+        replicationPluginInterface.stopReplication(client, request, listener)
+        // Verify that listener.onResponse is called with the correct response
+        verify(listener).onFailure(exception)
     }
 }