Handle serialization issues with UpdateReplicationStateDetailsRequest (…

…opensearch-project#866)

Signed-off-by: Ankit Kala <[email protected]>

src/main/kotlin/com/amazon/elasticsearch/replication/action/replicationstatedetails/UpdateReplicationStateDetailsRequest.kt

@@ -21,6 +21,7 @@ import org.elasticsearch.action.support.master.AcknowledgedRequest
 import org.elasticsearch.common.io.stream.StreamInput
 import org.elasticsearch.common.io.stream.StreamOutput
 
+
 class UpdateReplicationStateDetailsRequest: AcknowledgedRequest<UpdateReplicationStateDetailsRequest> {
 
     val followIndexName : String
@@ -52,6 +53,7 @@ class UpdateReplicationStateDetailsRequest: AcknowledgedRequest<UpdateReplicatio
     override fun writeTo(out: StreamOutput) {
         super.writeTo(out)
         out.writeString(followIndexName)
-        out.writeMap(replicationStateParams)
+        out.writeMap(replicationStateParams, StreamOutput::writeString, StreamOutput::writeString)
+        out.writeEnum(updateType)
     }
 }

src/test/kotlin/com/amazon/elasticsearch/replication/action/replicationstatedetails/UpdateReplicationStateDetailsRequestTests.kt

@@ -0,0 +1,42 @@
+package com.amazon.elasticsearch.replication.action.replicationstatedetails
+
+import org.assertj.core.api.Assertions
+import org.elasticsearch.common.io.stream.BytesStreamOutput
+import org.elasticsearch.test.ESTestCase
+
+class UpdateReplicationStateDetailsRequestTests: ESTestCase() {
+    companion object {
+        const val INDEX = "index"
+    }
+
+    fun `test serialization update type add`() {
+        val state = mapOf(Pair("k1", "v1"), Pair("k2", "v2"))
+        val request = UpdateReplicationStateDetailsRequest(INDEX, state, UpdateReplicationStateDetailsRequest.UpdateType.ADD)
+        val output = BytesStreamOutput()
+        request.writeTo(output)
+        val deserialized = UpdateReplicationStateDetailsRequest(output.bytes().streamInput())
+
+        assertEquals(UpdateReplicationStateDetailsRequest.UpdateType.ADD, deserialized.updateType)
+        assertEquals(INDEX, deserialized.followIndexName)
+        Assertions.assertThat(deserialized.replicationStateParams.containsKey("k1"))
+        Assertions.assertThat(deserialized.replicationStateParams.containsKey("k2"))
+        Assertions.assertThat("v1".equals(deserialized.replicationStateParams["k1"]))
+        Assertions.assertThat("v2".equals(deserialized.replicationStateParams["k2"]))
+    }
+
+    fun `test serialization update type delete`() {
+        val state = mapOf(Pair("k1", "v1"), Pair("k2", "v2"))
+        val request = UpdateReplicationStateDetailsRequest(INDEX, state, UpdateReplicationStateDetailsRequest.UpdateType.REMOVE)
+        val output = BytesStreamOutput()
+        request.writeTo(output)
+
+        val deserialized = UpdateReplicationStateDetailsRequest(output.bytes().streamInput())
+
+        assertEquals(UpdateReplicationStateDetailsRequest.UpdateType.REMOVE, deserialized.updateType)
+        assertEquals(INDEX, deserialized.followIndexName)
+        Assertions.assertThat(deserialized.replicationStateParams.containsKey("k1"))
+        Assertions.assertThat(deserialized.replicationStateParams.containsKey("k2"))
+        Assertions.assertThat("v1".equals(deserialized.replicationStateParams["k1"]))
+        Assertions.assertThat("v2".equals(deserialized.replicationStateParams["k2"]))
+    }
+}