initialize serde failed flag to false

Signed-off-by: Surya Sashank Nistala <[email protected]>

Author: eirsep

src/main/kotlin/org/opensearch/commons/alerting/action/DocLevelMonitorFanOutRequest.kt

@@ -40,6 +40,9 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
     val workflowRunContext: WorkflowRunContext?
     val hasSerializationFailed: Boolean
 
+    init {
+        serializationFailedFlag = false
+    }
     companion object {
         // flag flipped to true whenever a safeRead*() method fails to serialize a field correctly
         private var serializationFailedFlag: Boolean = false
@@ -99,7 +102,7 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             } catch (e: Exception) {
                 serializationFailedFlag = true
                 log.error("Error parsing shardId list in Doc level monitor fanout request", e)
-                listOf(ShardId("failed_serde", "failed_serde", 999999))
+                emptyList()
             }
 
         private fun safeReadStringList(sin: StreamInput): List<String> =
@@ -124,7 +127,8 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             var indexExecutionContext: IndexExecutionContext? = null
             return try {
                 indexExecutionContext = IndexExecutionContext(sin)
-                while (sin.read() != -1) {
+                while (sin.read() != 0) {
+                    serializationFailedFlag = true
                     // read and discard bytes until stream is entirely consumed
                     try {
                         sin.readByte()
@@ -137,7 +141,7 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
             } catch (e: Exception) {
                 serializationFailedFlag = true
                 log.error("Error parsing index execution context in Doc level monitor fanout request", e)
-                while (sin.read() != -1) {
+                while (sin.read() != 0) {
                     try { // read and throw bytes until stream is entirely consumed
                         sin.readByte()
                     } catch (_: EOFException) {
@@ -148,7 +152,7 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
         }
     }
 
-    constructor(
+    private constructor(
         monitor: Monitor,
         dryRun: Boolean,
         monitorMetadata: MonitorMetadata,
@@ -157,7 +161,7 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
         shardIds: List<ShardId>,
         concreteIndicesSeenSoFar: List<String>,
         workflowRunContext: WorkflowRunContext?,
-        hasSerializationFailed: Boolean? = null
+        hasSerializationFailed: Boolean
     ) : super() {
         this.monitor = monitor
         this.dryRun = dryRun
@@ -167,10 +171,30 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
         this.shardIds = shardIds
         this.concreteIndicesSeenSoFar = concreteIndicesSeenSoFar
         this.workflowRunContext = workflowRunContext
-        require(false == shardIds.isEmpty()) { }
         this.hasSerializationFailed = hasSerializationFailed ?: false
     }
 
+    constructor(
+        monitor: Monitor,
+        dryRun: Boolean,
+        monitorMetadata: MonitorMetadata,
+        executionId: String,
+        indexExecutionContext: IndexExecutionContext?,
+        shardIds: List<ShardId>,
+        concreteIndicesSeenSoFar: List<String>,
+        workflowRunContext: WorkflowRunContext?
+    ) : super() {
+        this.monitor = monitor
+        this.dryRun = dryRun
+        this.monitorMetadata = monitorMetadata
+        this.executionId = executionId
+        this.indexExecutionContext = indexExecutionContext
+        this.shardIds = shardIds
+        this.concreteIndicesSeenSoFar = concreteIndicesSeenSoFar
+        this.workflowRunContext = workflowRunContext
+        this.hasSerializationFailed = false
+    }
+
     @Throws(IOException::class)
     constructor(sin: StreamInput) : this(
         monitor = safeReadMonitor(sin),

src/test/kotlin/org/opensearch/commons/alerting/action/DocLevelMonitorFanOutRequestTests.kt

@@ -91,7 +91,7 @@ class DocLevelMonitorFanOutRequestTests {
         assertEquals(docLevelMonitorFanOutRequest.indexExecutionContext, newDocLevelMonitorFanOutRequest.indexExecutionContext)
         assertEquals(docLevelMonitorFanOutRequest.shardIds, newDocLevelMonitorFanOutRequest.shardIds)
         assertEquals(docLevelMonitorFanOutRequest.workflowRunContext, newDocLevelMonitorFanOutRequest.workflowRunContext)
-        assertEquals(sin.read(), -1)
+        assertEquals(sin.read(), 0)
         assertFalse(newDocLevelMonitorFanOutRequest.hasSerializationFailed)
     }
 
@@ -156,7 +156,7 @@ class DocLevelMonitorFanOutRequestTests {
         assertEquals(docLevelMonitorFanOutRequest.shardIds, newDocLevelMonitorFanOutRequest.shardIds)
         assertEquals(docLevelMonitorFanOutRequest.workflowRunContext, newDocLevelMonitorFanOutRequest.workflowRunContext)
         assertFalse(newDocLevelMonitorFanOutRequest.hasSerializationFailed)
-        assertEquals(sin.read(), -1)
+        assertEquals(sin.read().toString(), sin.read(), 0)
     }
 
     @Test
@@ -217,4 +217,68 @@ class DocLevelMonitorFanOutRequestTests {
         assertTrue(newDocLevelMonitorFanOutRequest.hasSerializationFailed)
         assertEquals(sin.read(), -1)
     }
+
+    @Test
+    fun `test doc level monitor fan out request as stream when there are additional bytes left to handle`() {
+        val docQuery = DocLevelQuery(query = "test_field:\"us-west-2\"", fields = listOf(), name = "3")
+        val docLevelInput = DocLevelMonitorInput("description", listOf("test-index"), listOf(docQuery))
+
+        val trigger = randomDocumentLevelTrigger(condition = Script("return true"))
+        val monitor = randomDocumentLevelMonitor(
+            inputs = listOf(docLevelInput),
+            triggers = listOf(trigger),
+            enabled = true,
+            schedule = IntervalSchedule(1, ChronoUnit.MINUTES)
+        )
+        val monitorMetadata = MonitorMetadata(
+            "test",
+            SequenceNumbers.UNASSIGNED_SEQ_NO,
+            SequenceNumbers.UNASSIGNED_PRIMARY_TERM,
+            Monitor.NO_ID,
+            listOf(ActionExecutionTime("", Instant.now())),
+            mutableMapOf("index" to mutableMapOf("1" to "1")),
+            mutableMapOf("test-index" to ".opensearch-sap-test_windows-queries-000001")
+        )
+        val indexExecutionContext = IndexExecutionContext(
+            listOf(docQuery),
+            mutableMapOf("index" to mutableMapOf("1" to "1")),
+            mutableMapOf("index" to mutableMapOf("1" to "1")),
+            "test-index",
+            "test-index",
+            listOf("test-index"),
+            listOf("test-index"),
+            listOf("test-field"),
+            listOf("1", "2")
+        )
+        val workflowRunContext = WorkflowRunContext(
+            Workflow.NO_ID,
+            Workflow.NO_ID,
+            Monitor.NO_ID,
+            mutableMapOf("index" to listOf("1")),
+            true
+        )
+        val docLevelMonitorFanOutRequest = DocLevelMonitorFanOutRequest(
+            monitor,
+            false,
+            monitorMetadata,
+            UUID.randomUUID().toString(),
+            indexExecutionContext,
+            listOf(ShardId("test-index", UUID.randomUUID().toString(), 0)),
+            listOf("test-index"),
+            workflowRunContext
+        )
+        val out = BytesStreamOutput()
+        docLevelMonitorFanOutRequest.writeTo(out)
+        out.writeByte(Byte.MIN_VALUE)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newDocLevelMonitorFanOutRequest = DocLevelMonitorFanOutRequest(sin)
+        assertEquals(docLevelMonitorFanOutRequest.monitor, newDocLevelMonitorFanOutRequest.monitor)
+        assertEquals(docLevelMonitorFanOutRequest.executionId, newDocLevelMonitorFanOutRequest.executionId)
+        assertEquals(docLevelMonitorFanOutRequest.monitorMetadata, newDocLevelMonitorFanOutRequest.monitorMetadata)
+        assertEquals(docLevelMonitorFanOutRequest.indexExecutionContext, newDocLevelMonitorFanOutRequest.indexExecutionContext)
+        assertEquals(docLevelMonitorFanOutRequest.shardIds, newDocLevelMonitorFanOutRequest.shardIds)
+        assertEquals(docLevelMonitorFanOutRequest.workflowRunContext, newDocLevelMonitorFanOutRequest.workflowRunContext)
+        assertEquals(sin.read(), 0)
+        assertTrue(newDocLevelMonitorFanOutRequest.hasSerializationFailed)
+    }
 }