force consume stream in fan out request

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

Author: eirsep

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

@@ -5,6 +5,7 @@
 
 package org.opensearch.commons.alerting.action
 
+import org.apache.logging.log4j.LogManager
 import org.opensearch.action.ActionRequest
 import org.opensearch.action.ActionRequestValidationException
 import org.opensearch.commons.alerting.model.DataSources
@@ -22,11 +23,13 @@ import org.opensearch.core.xcontent.ToXContent
 import org.opensearch.core.xcontent.ToXContentObject
 import org.opensearch.core.xcontent.XContentBuilder
 import org.opensearch.index.seqno.SequenceNumbers
+import java.io.EOFException
 import java.io.IOException
 import java.time.Instant
 import java.time.temporal.ChronoUnit
 
 class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
+    private val log = LogManager.getLogger(DocLevelMonitorFanOutRequest::class.java)
     val monitor: Monitor
     val dryRun: Boolean
     val monitorMetadata: MonitorMetadata
@@ -99,12 +102,30 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
                 null
             }
 
-        private fun safeReadIndexExecutionContext(sin: StreamInput): IndexExecutionContext? =
-            try {
-                IndexExecutionContext(sin)
+        private fun safeReadIndexExecutionContext(sin: StreamInput): IndexExecutionContext? {
+            var indexExecutionContext: IndexExecutionContext? = null
+            return try {
+                indexExecutionContext = IndexExecutionContext(sin)
+                while (sin.read() != -1) {
+                    try { // read and throw bytes until stream is entirely consumed
+                        sin.readByte()
+                    } catch (_: EOFException) {
+                    }
+                }
+                return indexExecutionContext
+            } catch (e: EOFException) {
+                indexExecutionContext
+
             } catch (e: Exception) {
+                while (sin.read() != -1) {
+                    try { // read and throw bytes until stream is entirely consumed
+                        sin.readByte()
+                    } catch (_: EOFException) {
+                    }
+                }
                 null
             }
+        }
     }
 
     constructor(
@@ -151,6 +172,7 @@ class DocLevelMonitorFanOutRequest : ActionRequest, ToXContentObject {
         out.writeBoolean(workflowRunContext != null)
         workflowRunContext?.writeTo(out)
         indexExecutionContext?.writeTo(out)
+        indexExecutionContext?.writeTo(out)
     }
 
     override fun validate(): ActionRequestValidationException? {

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

@@ -7,7 +7,6 @@ package org.opensearch.commons.alerting.action
 
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNull
-import org.junit.Assert.assertTrue
 import org.junit.jupiter.api.Test
 import org.opensearch.common.io.stream.BytesStreamOutput
 import org.opensearch.commons.alerting.model.ActionExecutionTime
@@ -82,6 +81,7 @@ class DocLevelMonitorFanOutRequestTests {
         )
         val out = BytesStreamOutput()
         docLevelMonitorFanOutRequest.writeTo(out)
+        monitor.writeTo(out)
         val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
         val newDocLevelMonitorFanOutRequest = DocLevelMonitorFanOutRequest(sin)
         assertEquals(docLevelMonitorFanOutRequest.monitor, newDocLevelMonitorFanOutRequest.monitor)