From 6d3e68a22d75327bf169dd32dfbbbeb7c38b4a09 Mon Sep 17 00:00:00 2001 From: thecodingshrimp Date: Wed, 19 Aug 2026 16:33:24 +0200 Subject: [PATCH] Fix immutable map ClassCastException bugs Signed-off-by: thecodingshrimp --- .../action/DocLevelMonitorFanOutResponse.kt | 13 +- .../BucketSelectorExtAggregationBuilder.kt | 3 +- .../commons/alerting/model/Alert.kt | 4 +- .../model/BucketLevelTriggerRunResult.kt | 3 +- .../model/ChainedAlertTriggerRunResult.kt | 3 +- .../model/ClusterMetricsTriggerRunResult.kt | 3 +- .../alerting/model/IndexExecutionContext.kt | 5 +- .../commons/alerting/model/Monitor.kt | 12 +- .../commons/alerting/model/MonitorMetadata.kt | 15 +- .../alerting/model/MonitorRunResult.kt | 25 +- .../model/QueryLevelTriggerRunResult.kt | 3 +- .../alerting/model/TriggerRunResult.kt | 7 - .../commons/alerting/model/Workflow.kt | 5 - .../alerting/model/WorkflowRunResult.kt | 8 +- .../alerting/util/StreamInputExtensions.kt | 16 + .../ImmutableMapDeserializationTests.kt | 719 ++++++++++++++++++ 16 files changed, 778 insertions(+), 66 deletions(-) create mode 100644 src/main/kotlin/org/opensearch/commons/alerting/util/StreamInputExtensions.kt create mode 100644 src/test/kotlin/org/opensearch/commons/alerting/ImmutableMapDeserializationTests.kt diff --git a/src/main/kotlin/org/opensearch/commons/alerting/action/DocLevelMonitorFanOutResponse.kt b/src/main/kotlin/org/opensearch/commons/alerting/action/DocLevelMonitorFanOutResponse.kt index 6e5cde551..04ce18ce0 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/action/DocLevelMonitorFanOutResponse.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/action/DocLevelMonitorFanOutResponse.kt @@ -8,6 +8,7 @@ package org.opensearch.commons.alerting.action import org.opensearch.commons.alerting.model.DocumentLevelTriggerRunResult import org.opensearch.commons.alerting.model.InputRunResults import org.opensearch.commons.alerting.util.AlertingException +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.action.ActionResponse import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput @@ -30,9 +31,9 @@ class DocLevelMonitorFanOutResponse : ActionResponse, ToXContentObject { nodeId = sin.readString(), executionId = sin.readString(), monitorId = sin.readString(), - lastRunContexts = sin.readMap()!! as MutableMap, + lastRunContexts = sin.readMapAsMutableMap() as MutableMap, inputResults = InputRunResults.readFrom(sin), - triggerResults = suppressWarning(sin.readMap(StreamInput::readString, DocumentLevelTriggerRunResult::readFrom)), + triggerResults = readTriggerResults(sin), exception = sin.readException() ) @@ -84,9 +85,11 @@ class DocLevelMonitorFanOutResponse : ActionResponse, ToXContentObject { } companion object { - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): Map { - return map as Map + private fun readTriggerResults(sin: StreamInput): Map { + val raw = sin.readMap(StreamInput::readString, DocumentLevelTriggerRunResult::readFrom) + if (raw.isEmpty()) return mutableMapOf() + @Suppress("UNCHECKED_CAST") + return HashMap(raw as Map) } } } diff --git a/src/main/kotlin/org/opensearch/commons/alerting/aggregation/bucketselectorext/BucketSelectorExtAggregationBuilder.kt b/src/main/kotlin/org/opensearch/commons/alerting/aggregation/bucketselectorext/BucketSelectorExtAggregationBuilder.kt index 12a09e1fc..563ac9ca3 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/aggregation/bucketselectorext/BucketSelectorExtAggregationBuilder.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/aggregation/bucketselectorext/BucketSelectorExtAggregationBuilder.kt @@ -42,7 +42,8 @@ class BucketSelectorExtAggregationBuilder : @Throws(IOException::class) @Suppress("UNCHECKED_CAST") constructor(sin: StreamInput) : super(sin, NAME.preferredName) { - bucketsPathsMap = sin.readMap() as MutableMap + @Suppress("UNCHECKED_CAST") + bucketsPathsMap = (sin.readMap()?.toMutableMap() ?: mutableMapOf()) as Map script = Script(sin) gapPolicy = BucketHelpers.GapPolicy.readFrom(sin) parentBucketPath = sin.readString() diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt index f0fa69c9a..45a30172a 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt @@ -3,11 +3,11 @@ package org.opensearch.commons.alerting.model import org.opensearch.Version import org.opensearch.common.lucene.uid.Versions import org.opensearch.commons.alerting.alerts.AlertError -import org.opensearch.commons.alerting.model.Monitor.Companion.suppressWarning import org.opensearch.commons.alerting.util.IndexUtils.Companion.NO_SCHEMA_VERSION import org.opensearch.commons.alerting.util.instant import org.opensearch.commons.alerting.util.optionalTimeField import org.opensearch.commons.alerting.util.optionalUserField +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.commons.authuser.User import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput @@ -423,7 +423,7 @@ data class Alert( null }, queryResults = if (sin.version.onOrAfter(Version.V_3_7_0)) { - sin.readList { input -> suppressWarning(input.readMap()) } + sin.readList { input -> input.readMapAsMutableMap() } } else { listOf() }, diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/BucketLevelTriggerRunResult.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/BucketLevelTriggerRunResult.kt index 34328ca23..c67e86a25 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/BucketLevelTriggerRunResult.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/BucketLevelTriggerRunResult.kt @@ -5,6 +5,7 @@ package org.opensearch.commons.alerting.model +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.xcontent.ToXContent @@ -24,7 +25,7 @@ data class BucketLevelTriggerRunResult( sin.readString(), sin.readException() as Exception?, // error sin.readMap(StreamInput::readString, ::AggregationResultBucket), - sin.readMap() as MutableMap> + sin.readMapAsMutableMap() as MutableMap> ) override fun internalXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/ChainedAlertTriggerRunResult.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/ChainedAlertTriggerRunResult.kt index 015762cf7..4c0838ef9 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/ChainedAlertTriggerRunResult.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/ChainedAlertTriggerRunResult.kt @@ -6,6 +6,7 @@ package org.opensearch.commons.alerting.model import org.opensearch.commons.alerting.alerts.AlertError +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.xcontent.ToXContent @@ -28,7 +29,7 @@ data class ChainedAlertTriggerRunResult( triggerName = sin.readString(), error = sin.readException(), triggered = sin.readBoolean(), - actionResults = sin.readMap() as MutableMap, + actionResults = sin.readMapAsMutableMap() as MutableMap, associatedAlertIds = sin.readStringList().toSet() ) diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/ClusterMetricsTriggerRunResult.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/ClusterMetricsTriggerRunResult.kt index d3af9be3e..54e636306 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/ClusterMetricsTriggerRunResult.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/ClusterMetricsTriggerRunResult.kt @@ -6,6 +6,7 @@ package org.opensearch.commons.alerting.model import org.opensearch.commons.alerting.alerts.AlertError +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.common.io.stream.Writeable @@ -35,7 +36,7 @@ data class ClusterMetricsTriggerRunResult( triggerName = sin.readString(), error = sin.readException(), triggered = sin.readBoolean(), - actionResults = sin.readMap() as MutableMap, + actionResults = sin.readMapAsMutableMap() as MutableMap, clusterTriggerResults = sin.readList((ClusterTriggerResult)::readFrom) ) diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/IndexExecutionContext.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/IndexExecutionContext.kt index 6b104d135..c932eeaec 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/IndexExecutionContext.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/IndexExecutionContext.kt @@ -6,6 +6,7 @@ package org.opensearch.commons.alerting.model import org.opensearch.Version +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.common.io.stream.Writeable @@ -29,8 +30,8 @@ data class IndexExecutionContext( @Throws(IOException::class) constructor(sin: StreamInput) : this( queries = sin.readList { DocLevelQuery(sin) }, - lastRunContext = sin.readMap() as MutableMap, - updatedLastRunContext = sin.readMap() as MutableMap, + lastRunContext = sin.readMapAsMutableMap() as MutableMap, + updatedLastRunContext = sin.readMapAsMutableMap() as MutableMap, indexName = sin.readString(), concreteIndexName = sin.readString(), updatedIndexNames = sin.readStringList(), diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/Monitor.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/Monitor.kt index d26b2b9d0..cbe9fe4c3 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/Monitor.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/Monitor.kt @@ -13,6 +13,7 @@ import org.opensearch.commons.alerting.util.isBucketLevelMonitor import org.opensearch.commons.alerting.util.isPPLMonitor import org.opensearch.commons.alerting.util.optionalTimeField import org.opensearch.commons.alerting.util.optionalUserField +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.commons.authuser.User import org.opensearch.core.ParseField import org.opensearch.core.common.io.stream.StreamInput @@ -131,7 +132,7 @@ data class Monitor( schemaVersion = sin.readInt(), inputs = sin.readList((Input)::readFrom), triggers = sin.readList((Trigger)::readFrom), - uiMetadata = suppressWarning(sin.readMap()), + uiMetadata = sin.readMapAsMutableMap(), dataSources = if (sin.readBoolean()) { DataSources(sin) } else { @@ -350,7 +351,7 @@ data class Monitor( var schedule: Schedule? = null var lastUpdateTime: Instant? = null var enabledTime: Instant? = null - var uiMetadata: Map = mapOf() + var uiMetadata: Map = mutableMapOf() var enabled = true var schemaVersion = NO_SCHEMA_VERSION val triggers: MutableList = mutableListOf() @@ -407,7 +408,7 @@ data class Monitor( } ENABLED_TIME_FIELD -> enabledTime = xcp.instant() LAST_UPDATE_TIME_FIELD -> lastUpdateTime = xcp.instant() - UI_METADATA_FIELD -> uiMetadata = xcp.map() + UI_METADATA_FIELD -> uiMetadata = xcp.map().toMutableMap() DATA_SOURCES_FIELD -> dataSources = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) { DataSources() } else { @@ -480,10 +481,5 @@ data class Monitor( fun readFrom(sin: StreamInput): Monitor? { return Monitor(sin) } - - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): MutableMap { - return map as MutableMap - } } } diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorMetadata.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorMetadata.kt index a90f3cc3f..ee6353a08 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorMetadata.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorMetadata.kt @@ -7,6 +7,7 @@ package org.opensearch.commons.alerting.model import org.opensearch.commons.alerting.model.Monitor.Companion.NO_ID import org.opensearch.commons.alerting.util.instant +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.common.io.stream.Writeable @@ -36,8 +37,8 @@ data class MonitorMetadata( primaryTerm = sin.readLong(), monitorId = sin.readString(), lastActionExecutionTimes = sin.readList(ActionExecutionTime.Companion::readFrom), - lastRunContext = Monitor.suppressWarning(sin.readMap()), - sourceToQueryIndexMapping = sin.readMap() as MutableMap + lastRunContext = sin.readMapAsMutableMap(), + sourceToQueryIndexMapping = sin.readMapAsMutableMap() as MutableMap ) override fun writeTo(out: StreamOutput) { @@ -47,7 +48,7 @@ data class MonitorMetadata( out.writeString(monitorId) out.writeCollection(lastActionExecutionTimes) out.writeMap(lastRunContext) - out.writeMap(sourceToQueryIndexMapping as MutableMap) + out.writeMap(sourceToQueryIndexMapping as Map) } override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { @@ -57,7 +58,7 @@ data class MonitorMetadata( .field(LAST_ACTION_EXECUTION_FIELD, lastActionExecutionTimes.toTypedArray()) if (lastRunContext.isNotEmpty()) builder.field(LAST_RUN_CONTEXT_FIELD, lastRunContext) if (sourceToQueryIndexMapping.isNotEmpty()) { - builder.field(SOURCE_TO_QUERY_INDEX_MAP_FIELD, sourceToQueryIndexMapping as MutableMap) + builder.field(SOURCE_TO_QUERY_INDEX_MAP_FIELD, sourceToQueryIndexMapping as Map) } if (params.paramAsBoolean("with_type", false)) builder.endObject() return builder.endObject() @@ -81,7 +82,7 @@ data class MonitorMetadata( ): MonitorMetadata { lateinit var monitorId: String val lastActionExecutionTimes = mutableListOf() - var lastRunContext: Map = mapOf() + var lastRunContext: Map = mutableMapOf() var sourceToQueryIndexMapping: MutableMap = mutableMapOf() XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp) @@ -97,8 +98,8 @@ data class MonitorMetadata( lastActionExecutionTimes.add(ActionExecutionTime.parse(xcp)) } } - LAST_RUN_CONTEXT_FIELD -> lastRunContext = xcp.map() - SOURCE_TO_QUERY_INDEX_MAP_FIELD -> sourceToQueryIndexMapping = xcp.map() as MutableMap + LAST_RUN_CONTEXT_FIELD -> lastRunContext = xcp.map().toMutableMap() + SOURCE_TO_QUERY_INDEX_MAP_FIELD -> sourceToQueryIndexMapping = (xcp.map()?.toMutableMap() ?: mutableMapOf()) as MutableMap } } diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorRunResult.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorRunResult.kt index c81253949..84c52d56f 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorRunResult.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/MonitorRunResult.kt @@ -10,6 +10,7 @@ import org.opensearch.OpenSearchException import org.opensearch.Version import org.opensearch.commons.alerting.alerts.AlertError import org.opensearch.commons.alerting.util.optionalTimeField +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.common.io.stream.Writeable @@ -36,7 +37,7 @@ data class MonitorRunResult( sin.readInstant(), // periodEnd sin.readException(), // error InputRunResults.readFrom(sin), // inputResults - suppressWarning(sin.readMap()) as Map // triggerResults + sin.readMapAsMutableMap() as Map // triggerResults ) override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { @@ -72,11 +73,6 @@ data class MonitorRunResult( fun readFrom(sin: StreamInput): MonitorRunResult { return MonitorRunResult(sin) } - - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): Map { - return map as Map - } } @Throws(IOException::class) @@ -147,7 +143,7 @@ data class InputRunResults( val count = sin.readVInt() // count val list = mutableListOf>() for (i in 0 until count) { - list.add(suppressWarning(sin.readMap())) // result(map) + list.add(sin.readMapAsMutableMap()) // result(map) } val pplCount = if (sin.version.onOrAfter(Version.V_3_7_0)) { sin.readVInt() @@ -157,7 +153,7 @@ data class InputRunResults( val pplList = mutableListOf>() if (sin.version.onOrAfter(Version.V_3_7_0)) { for (i in 0 until pplCount) { - pplList.add(suppressWarning(sin.readMap())) // pplResults + pplList.add(sin.readMapAsMutableMap()) // pplResults } } val pplNumResults = if (sin.version.onOrAfter(Version.V_3_7_0)) { @@ -168,11 +164,6 @@ data class InputRunResults( val error = sin.readException() // error return InputRunResults(list, error, null, pplList, pplNumResults) } - - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): Map { - return map as Map - } } fun afterKeysPresent(): Boolean { @@ -196,11 +187,12 @@ data class ActionRunResult( val error: Exception? = null ) : Writeable, ToXContent { + @Suppress("UNCHECKED_CAST") @Throws(IOException::class) constructor(sin: StreamInput) : this( sin.readString(), // actionId sin.readString(), // actionName - suppressWarning(sin.readMap()), // output + sin.readMapAsMutableMap() as Map, // output sin.readBoolean(), // throttled sin.readOptionalInstant(), // executionTime sin.readException() // error @@ -233,11 +225,6 @@ data class ActionRunResult( fun readFrom(sin: StreamInput): ActionRunResult { return ActionRunResult(sin) } - - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): MutableMap { - return map as MutableMap - } } } diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/QueryLevelTriggerRunResult.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/QueryLevelTriggerRunResult.kt index 04e775646..5aad0d5f6 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/QueryLevelTriggerRunResult.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/QueryLevelTriggerRunResult.kt @@ -7,6 +7,7 @@ package org.opensearch.commons.alerting.model import org.opensearch.Version import org.opensearch.commons.alerting.alerts.AlertError +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.xcontent.ToXContent @@ -29,7 +30,7 @@ open class QueryLevelTriggerRunResult( triggerName = sin.readString(), error = sin.readException(), triggered = sin.readBoolean(), - actionResults = sin.readMap() as MutableMap, + actionResults = sin.readMapAsMutableMap() as MutableMap, pplCustomQueryResults = if (sin.version.onOrAfter(Version.V_3_7_0)) { sin.readList { it.readMap() } } else { diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/TriggerRunResult.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/TriggerRunResult.kt index 84efde395..0113594dc 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/TriggerRunResult.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/TriggerRunResult.kt @@ -45,11 +45,4 @@ abstract class TriggerRunResult( out.writeString(triggerName) out.writeException(error) } - - companion object { - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): MutableMap { - return map as MutableMap - } - } } diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/Workflow.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/Workflow.kt index d0e57e63e..f26174f79 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/Workflow.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/Workflow.kt @@ -294,11 +294,6 @@ data class Workflow( return Workflow(sin) } - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): MutableMap { - return map as MutableMap - } - private const val DEFAULT_OWNER = "alerting" } } diff --git a/src/main/kotlin/org/opensearch/commons/alerting/model/WorkflowRunResult.kt b/src/main/kotlin/org/opensearch/commons/alerting/model/WorkflowRunResult.kt index 1b5fe3d82..e49bb7aca 100644 --- a/src/main/kotlin/org/opensearch/commons/alerting/model/WorkflowRunResult.kt +++ b/src/main/kotlin/org/opensearch/commons/alerting/model/WorkflowRunResult.kt @@ -5,6 +5,7 @@ package org.opensearch.commons.alerting.model +import org.opensearch.commons.alerting.util.readMapAsMutableMap import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import org.opensearch.core.common.io.stream.Writeable @@ -35,7 +36,7 @@ data class WorkflowRunResult( executionEndTime = sin.readOptionalInstant(), executionId = sin.readString(), error = sin.readException(), - triggerResults = suppressWarning(sin.readMap()) as Map + triggerResults = sin.readMapAsMutableMap() as Map ) override fun writeTo(out: StreamOutput) { @@ -73,10 +74,5 @@ data class WorkflowRunResult( fun readFrom(sin: StreamInput): WorkflowRunResult { return WorkflowRunResult(sin) } - - @Suppress("UNCHECKED_CAST") - fun suppressWarning(map: MutableMap?): Map { - return map as Map - } } } diff --git a/src/main/kotlin/org/opensearch/commons/alerting/util/StreamInputExtensions.kt b/src/main/kotlin/org/opensearch/commons/alerting/util/StreamInputExtensions.kt new file mode 100644 index 000000000..efc6943e9 --- /dev/null +++ b/src/main/kotlin/org/opensearch/commons/alerting/util/StreamInputExtensions.kt @@ -0,0 +1,16 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.commons.alerting.util + +import org.opensearch.core.common.io.stream.StreamInput + +/** + * Reads a map from StreamInput and guarantees a mutable result. + * StreamInput.readMap() returns Collections.emptyMap() (immutable) for zero-size maps. + * This extension safely converts that to a mutable LinkedHashMap. + */ +fun StreamInput.readMapAsMutableMap(): MutableMap = + readMap()?.toMutableMap() ?: mutableMapOf() diff --git a/src/test/kotlin/org/opensearch/commons/alerting/ImmutableMapDeserializationTests.kt b/src/test/kotlin/org/opensearch/commons/alerting/ImmutableMapDeserializationTests.kt new file mode 100644 index 000000000..cf7b425d0 --- /dev/null +++ b/src/test/kotlin/org/opensearch/commons/alerting/ImmutableMapDeserializationTests.kt @@ -0,0 +1,719 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.commons.alerting + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.opensearch.common.io.stream.BytesStreamOutput +import org.opensearch.common.settings.Settings +import org.opensearch.commons.alerting.action.DocLevelMonitorFanOutResponse +import org.opensearch.commons.alerting.aggregation.bucketselectorext.BucketSelectorExtAggregationBuilder +import org.opensearch.commons.alerting.model.ActionExecutionTime +import org.opensearch.commons.alerting.model.ActionRunResult +import org.opensearch.commons.alerting.model.AggregationResultBucket +import org.opensearch.commons.alerting.model.Alert +import org.opensearch.commons.alerting.model.BucketLevelTriggerRunResult +import org.opensearch.commons.alerting.model.ChainedAlertTriggerRunResult +import org.opensearch.commons.alerting.model.ClusterMetricsTriggerRunResult +import org.opensearch.commons.alerting.model.DocumentLevelTriggerRunResult +import org.opensearch.commons.alerting.model.IndexExecutionContext +import org.opensearch.commons.alerting.model.InputRunResults +import org.opensearch.commons.alerting.model.Monitor +import org.opensearch.commons.alerting.model.MonitorMetadata +import org.opensearch.commons.alerting.model.MonitorRunResult +import org.opensearch.commons.alerting.model.QueryLevelTriggerRunResult +import org.opensearch.commons.alerting.model.TriggerRunResult +import org.opensearch.commons.alerting.model.WorkflowRunResult +import org.opensearch.commons.alerting.util.getBucketKeysHash +import org.opensearch.core.common.io.stream.NamedWriteableAwareStreamInput +import org.opensearch.core.common.io.stream.NamedWriteableRegistry +import org.opensearch.core.common.io.stream.StreamInput +import org.opensearch.search.SearchModule +import java.time.Instant +import java.time.temporal.ChronoUnit + +/** + * Regression tests for StreamInput.readMap() immutable-map ClassCastException. + * + * StreamInput.readMap() Javadoc: "If the returned map contains any entries it will be + * mutable. If it is empty it might be immutable." — meaning Collections.emptyMap() is + * returned for zero-size maps. Any code that casts the result to MutableMap without a + * defensive copy throws ClassCastException (or UnsupportedOperationException on write) + * when the deserialized map is empty. + * + * Each test exercises the empty-map path: serialize an object with empty maps, deserialize + * it, then assert the round-trip succeeded and the result is mutable. + * + * See: opensearch-project/common-utils#967 + */ +class ImmutableMapDeserializationTests { + + // ------------------------------------------------------------------------- + // Monitor.uiMetadata + // ------------------------------------------------------------------------- + + @Test + fun `Monitor with empty uiMetadata survives serialization round-trip`() { + val monitor = randomQueryLevelMonitor(withMetadata = false) // uiMetadata = mapOf() + assertTrue(monitor.uiMetadata.isEmpty()) + + val out = BytesStreamOutput() + monitor.writeTo(out) + val registry = NamedWriteableRegistry(SearchModule(Settings.EMPTY, emptyList()).namedWriteables) + val sin = NamedWriteableAwareStreamInput(StreamInput.wrap(out.bytes().toBytesRef().bytes), registry) + val deserialized = Monitor(sin) + + assertEquals(monitor.name, deserialized.name) + assertTrue(deserialized.uiMetadata.isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException + assertMutableMap(deserialized.uiMetadata) + } + + @Test + fun `Monitor with non-empty uiMetadata survives serialization round-trip`() { + val monitor = randomQueryLevelMonitor(withMetadata = true) // uiMetadata = mapOf("foo" to "bar") + assertTrue(monitor.uiMetadata.isNotEmpty()) + + val out = BytesStreamOutput() + monitor.writeTo(out) + val registry = NamedWriteableRegistry(SearchModule(Settings.EMPTY, emptyList()).namedWriteables) + val sin = NamedWriteableAwareStreamInput(StreamInput.wrap(out.bytes().toBytesRef().bytes), registry) + val deserialized = Monitor(sin) + + assertEquals(monitor.uiMetadata, deserialized.uiMetadata) + } + + // ------------------------------------------------------------------------- + // MonitorMetadata.lastRunContext + sourceToQueryIndexMapping + // ------------------------------------------------------------------------- + + @Test + fun `MonitorMetadata with empty lastRunContext survives serialization round-trip`() { + val metadata = MonitorMetadata( + id = "monitor-id-metadata", + monitorId = "monitor-id", + lastActionExecutionTimes = emptyList(), + lastRunContext = emptyMap(), + sourceToQueryIndexMapping = mutableMapOf() + ) + + val out = BytesStreamOutput() + metadata.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = MonitorMetadata(sin) + + assertEquals(metadata.id, deserialized.id) + assertTrue(deserialized.lastRunContext.isEmpty()) + assertTrue(deserialized.sourceToQueryIndexMapping.isEmpty()) + // Verify mutability of lastRunContext + assertMutableMap(deserialized.lastRunContext) + // Verify mutability of sourceToQueryIndexMapping + deserialized.sourceToQueryIndexMapping["idx"] = "query-idx" + } + + @Test + fun `MonitorMetadata with non-empty maps survives serialization round-trip`() { + val metadata = MonitorMetadata( + id = "monitor-id-metadata", + monitorId = "monitor-id", + lastActionExecutionTimes = listOf( + ActionExecutionTime("action-1", Instant.now().truncatedTo(ChronoUnit.MILLIS)) + ), + lastRunContext = mapOf("index" to mapOf("0" to "1234")), + sourceToQueryIndexMapping = mutableMapOf("my-index-monitor-id" to "query-index") + ) + + val out = BytesStreamOutput() + metadata.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = MonitorMetadata(sin) + + assertEquals(metadata.lastRunContext, deserialized.lastRunContext) + assertEquals(metadata.sourceToQueryIndexMapping, deserialized.sourceToQueryIndexMapping) + } + + // ------------------------------------------------------------------------- + // MonitorMetadata.parse() via XContent — regression for immutable EmptyMap + // ------------------------------------------------------------------------- + + @Test + fun `MonitorMetadata parse without last_run_context field returns mutable lastRunContext`() { + // toXContent() omits last_run_context when empty, so a stored document has no such field. + // Before the fix, the default was mapOf() == kotlin.collections.EmptyMap (immutable). + // Any cast to MutableMap would throw ClassCastException. + val json = """ + { + "monitor_id": "test-monitor-id", + "last_action_execution_times": [] + } + """.trimIndent() + + val metadata = MonitorMetadata.parse(parser(json), id = "test-monitor-id-metadata") + + assertTrue(metadata.lastRunContext.isEmpty()) + assertMutableMap(metadata.lastRunContext) + } + + @Test + fun `MonitorMetadata parse with last_run_context field returns mutable lastRunContext`() { + // xcp.map() without .toMutableMap() can return an immutable map when the parsed map + // is backed by an immutable collection. Ensure the XContent assignment path is also safe. + val json = """ + { + "monitor_id": "test-monitor-id", + "last_action_execution_times": [], + "last_run_context": { + "my-index": {"0": "42", "1": "17"} + } + } + """.trimIndent() + + val metadata = MonitorMetadata.parse(parser(json), id = "test-monitor-id-metadata") + + assertEquals(1, metadata.lastRunContext.size) + assertMutableMap(metadata.lastRunContext) + } + + // ------------------------------------------------------------------------- + // Monitor.parse() via XContent — regression for immutable EmptyMap (uiMetadata) + // ------------------------------------------------------------------------- + + @Test + fun `Monitor parse without ui_metadata field returns mutable uiMetadata`() { + // toXContent() omits ui_metadata when empty; parse() default was mapOf() (immutable). + val json = """ + { + "name": "test-monitor", + "type": "monitor", + "monitor_type": "query_level_monitor", + "enabled": true, + "schedule": {"period": {"interval": 1, "unit": "MINUTES"}}, + "inputs": [], + "triggers": [], + "last_update_time": 0 + } + """.trimIndent() + + val monitor = Monitor.parse(parser(json)) + + assertTrue(monitor.uiMetadata.isEmpty()) + assertMutableMap(monitor.uiMetadata) + } + + @Test + fun `Monitor parse with ui_metadata field returns mutable uiMetadata`() { + val json = """ + { + "name": "test-monitor", + "type": "monitor", + "monitor_type": "query_level_monitor", + "enabled": true, + "schedule": {"period": {"interval": 1, "unit": "MINUTES"}}, + "inputs": [], + "triggers": [], + "last_update_time": 0, + "ui_metadata": {"layout": "graph", "version": 1} + } + """.trimIndent() + + val monitor = Monitor.parse(parser(json)) + + assertEquals(2, monitor.uiMetadata.size) + assertMutableMap(monitor.uiMetadata) + } + + // ------------------------------------------------------------------------- + // IndexExecutionContext.lastRunContext + updatedLastRunContext + // ------------------------------------------------------------------------- + + @Test + fun `IndexExecutionContext with empty lastRunContext survives serialization round-trip`() { + val ctx = IndexExecutionContext( + queries = emptyList(), + lastRunContext = mutableMapOf(), + updatedLastRunContext = mutableMapOf(), + indexName = "my-index", + concreteIndexName = "my-index-000001", + updatedIndexNames = emptyList(), + concreteIndexNames = emptyList(), + conflictingFields = emptyList(), + docIds = emptyList(), + findingIds = emptyList() + ) + + val out = BytesStreamOutput() + ctx.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = IndexExecutionContext(sin) + + assertEquals(ctx.indexName, deserialized.indexName) + assertTrue(deserialized.lastRunContext.isEmpty()) + assertTrue(deserialized.updatedLastRunContext.isEmpty()) + // Verify mutability + assertMutableMap(deserialized.lastRunContext) + assertMutableMap(deserialized.updatedLastRunContext) + } + + @Test + fun `IndexExecutionContext with non-empty context maps survives serialization round-trip`() { + val ctx = IndexExecutionContext( + queries = emptyList(), + lastRunContext = mutableMapOf("shard0" to "seq100"), + updatedLastRunContext = mutableMapOf("shard0" to "seq101"), + indexName = "my-index", + concreteIndexName = "my-index-000001", + updatedIndexNames = listOf("my-index"), + concreteIndexNames = listOf("my-index-000001"), + conflictingFields = emptyList() + ) + + val out = BytesStreamOutput() + ctx.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = IndexExecutionContext(sin) + + assertEquals(ctx.lastRunContext, deserialized.lastRunContext) + assertEquals(ctx.updatedLastRunContext, deserialized.updatedLastRunContext) + } + + // ------------------------------------------------------------------------- + // DocLevelMonitorFanOutResponse.lastRunContexts + // ------------------------------------------------------------------------- + + @Test + fun `DocLevelMonitorFanOutResponse with empty lastRunContexts survives serialization round-trip`() { + val response = DocLevelMonitorFanOutResponse( + nodeId = "node-1", + executionId = "exec-1", + monitorId = "monitor-1", + lastRunContexts = mutableMapOf(), // empty — triggers the bug + inputResults = InputRunResults() + ) + + val out = BytesStreamOutput() + response.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = DocLevelMonitorFanOutResponse(sin) + + assertEquals(response.nodeId, deserialized.nodeId) + assertEquals(response.executionId, deserialized.executionId) + assertTrue(deserialized.lastRunContexts.isEmpty()) + // Verify mutability + deserialized.lastRunContexts["index"] = mutableMapOf() + } + + @Test + fun `DocLevelMonitorFanOutResponse with empty triggerResults survives serialization round-trip`() { + val response = DocLevelMonitorFanOutResponse( + nodeId = "node-1", + executionId = "exec-1", + monitorId = "monitor-1", + lastRunContexts = mutableMapOf("index" to mutableMapOf() as Any), + inputResults = InputRunResults(), + triggerResults = mapOf() // empty — triggers the bug + ) + + val out = BytesStreamOutput() + response.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = DocLevelMonitorFanOutResponse(sin) + + assertEquals(response.nodeId, deserialized.nodeId) + assertTrue(deserialized.triggerResults.isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException or ClassCastException + @Suppress("UNCHECKED_CAST") + (deserialized.triggerResults as MutableMap) + .put("__mutable_check__", randomDocumentLevelTriggerRunResult()) + @Suppress("UNCHECKED_CAST") + (deserialized.triggerResults as MutableMap) + .remove("__mutable_check__") + } + + @Test + fun `DocLevelMonitorFanOutResponse with non-empty lastRunContexts survives serialization round-trip`() { + val response = DocLevelMonitorFanOutResponse( + nodeId = "node-1", + executionId = "exec-1", + monitorId = "monitor-1", + lastRunContexts = mutableMapOf("index" to mutableMapOf("0" to "100") as Any), + inputResults = InputRunResults(), + triggerResults = mapOf("t1" to randomDocumentLevelTriggerRunResult()) + ) + + val out = BytesStreamOutput() + response.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = DocLevelMonitorFanOutResponse(sin) + + assertEquals(response.nodeId, deserialized.nodeId) + assertEquals(response.lastRunContexts, deserialized.lastRunContexts) + } + + // ------------------------------------------------------------------------- + // WorkflowRunResult.triggerResults + // ------------------------------------------------------------------------- + + @Test + fun `WorkflowRunResult with empty triggerResults survives serialization round-trip`() { + val result = WorkflowRunResult( + workflowId = "workflow-1", + workflowName = "my-workflow", + monitorRunResults = emptyList(), + executionStartTime = Instant.now().truncatedTo(ChronoUnit.MILLIS), + executionEndTime = Instant.now().truncatedTo(ChronoUnit.MILLIS), + executionId = "exec-1", + triggerResults = mapOf() // empty — triggers the bug + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = WorkflowRunResult(sin) + + assertEquals(result.workflowId, deserialized.workflowId) + assertEquals(result.executionId, deserialized.executionId) + assertTrue(deserialized.triggerResults.isEmpty()) + } + + @Test + fun `WorkflowRunResult with non-empty triggerResults is correctly typed after deserialization`() { + // WorkflowRunResult.writeTo() uses the untyped writeMap which cannot serialize complex + // Writeable values like ChainedAlertTriggerRunResult. The existing production code path + // that populates triggerResults only uses the empty-map case at deserialization time + // (the map is populated by the caller after construction, not via StreamInput in practice). + // This test verifies the construction contract: a WorkflowRunResult with trigger results + // retains its data correctly when built directly. + val workflow = randomWorkflow() + val trigger = randomChainedAlertTrigger() + val triggerRunResult = ChainedAlertTriggerRunResult( + triggerName = trigger.name, + triggered = true, + error = null, + actionResults = mutableMapOf(), + associatedAlertIds = emptySet() + ) + val result = WorkflowRunResult( + workflowId = workflow.id, + workflowName = workflow.name, + monitorRunResults = emptyList(), + executionStartTime = Instant.now().truncatedTo(ChronoUnit.MILLIS), + executionEndTime = Instant.now().truncatedTo(ChronoUnit.MILLIS), + executionId = "exec-1", + triggerResults = mapOf(trigger.id to triggerRunResult) + ) + + assertEquals(1, result.triggerResults.size) + assertEquals(triggerRunResult, result.triggerResults[trigger.id]) + } + + // ------------------------------------------------------------------------- + // MonitorRunResult.triggerResults + // ------------------------------------------------------------------------- + + @Test + fun `MonitorRunResult with empty triggerResults survives serialization round-trip`() { + val result = MonitorRunResult( + monitorName = "test-monitor", + periodStart = Instant.now().truncatedTo(ChronoUnit.MILLIS), + periodEnd = Instant.now().truncatedTo(ChronoUnit.MILLIS), + triggerResults = mapOf() + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = MonitorRunResult(sin) + + assertEquals(result.monitorName, deserialized.monitorName) + assertTrue(deserialized.triggerResults.isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException + assertMutableMap(deserialized.triggerResults) + } + + @Test + fun `MonitorRunResult with non-empty triggerResults survives serialization round-trip`() { + val result = MonitorRunResult( + monitorName = "test-monitor", + periodStart = Instant.now().truncatedTo(ChronoUnit.MILLIS), + periodEnd = Instant.now().truncatedTo(ChronoUnit.MILLIS), + triggerResults = mapOf() + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = MonitorRunResult(sin) + + assertEquals(result.monitorName, deserialized.monitorName) + assertEquals(result.triggerResults, deserialized.triggerResults) + } + + // ------------------------------------------------------------------------- + // InputRunResults.results + // ------------------------------------------------------------------------- + + @Test + fun `InputRunResults with empty results list survives serialization round-trip`() { + val inputRunResults = InputRunResults( + results = listOf(), + error = null + ) + + val out = BytesStreamOutput() + inputRunResults.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = InputRunResults.readFrom(sin) + + assertTrue(deserialized.results.isEmpty()) + } + + @Test + fun `InputRunResults with results containing empty maps survives serialization round-trip`() { + val inputRunResults = InputRunResults( + results = listOf(emptyMap()), + error = null + ) + + val out = BytesStreamOutput() + inputRunResults.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = InputRunResults.readFrom(sin) + + assertEquals(1, deserialized.results.size) + assertTrue(deserialized.results[0].isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException + assertMutableMap(deserialized.results[0]) + } + + // ------------------------------------------------------------------------- + // Alert.queryResults (each element map) + // ------------------------------------------------------------------------- + + @Test + fun `Alert with empty queryResults list survives serialization round-trip`() { + val alert = randomAlert().copy(queryResults = emptyList()) + + val out = BytesStreamOutput() + alert.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = Alert(sin) + + assertEquals(alert.id, deserialized.id) + assertTrue(deserialized.queryResults.isEmpty()) + } + + @Test + fun `Alert with queryResults containing empty maps survives serialization round-trip`() { + // Each empty map in queryResults triggers readMap() → Collections.emptyMap() on deserialization + val alert = randomAlertWithPPLFields().copy( + queryResults = listOf(emptyMap(), emptyMap()) + ) + + val out = BytesStreamOutput() + alert.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = Alert(sin) + + assertEquals(2, deserialized.queryResults.size) + assertTrue(deserialized.queryResults[0].isEmpty()) + assertTrue(deserialized.queryResults[1].isEmpty()) + } + + @Test + fun `Alert with non-empty queryResults survives serialization round-trip`() { + val alert = randomAlertWithPPLFields() + assertTrue(alert.queryResults.isNotEmpty()) + + val out = BytesStreamOutput() + alert.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = Alert(sin) + + assertEquals(alert.queryResults, deserialized.queryResults) + } + + // ------------------------------------------------------------------------- + // QueryLevelTriggerRunResult.actionResults + // ------------------------------------------------------------------------- + + @Test + fun `QueryLevelTriggerRunResult with empty actionResults survives serialization round-trip`() { + val result = QueryLevelTriggerRunResult( + triggerName = "query-trigger", + triggered = false, + error = null, + actionResults = mutableMapOf() + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = QueryLevelTriggerRunResult(sin) + + assertEquals(result.triggerName, deserialized.triggerName) + assertTrue(deserialized.actionResults.isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException or ClassCastException + @Suppress("UNCHECKED_CAST") + (deserialized.actionResults as MutableMap) + .put("__mutable_check__", randomActionRunResult()) + @Suppress("UNCHECKED_CAST") + (deserialized.actionResults as MutableMap) + .remove("__mutable_check__") + } + + // ------------------------------------------------------------------------- + // ClusterMetricsTriggerRunResult.actionResults + // ------------------------------------------------------------------------- + + @Test + fun `ClusterMetricsTriggerRunResult with empty actionResults survives serialization round-trip`() { + val result = ClusterMetricsTriggerRunResult( + triggerName = "cluster-trigger", + triggered = false, + error = null, + actionResults = mutableMapOf(), + clusterTriggerResults = emptyList() + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = ClusterMetricsTriggerRunResult(sin) + + assertEquals(result.triggerName, deserialized.triggerName) + assertTrue(deserialized.actionResults.isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException or ClassCastException + @Suppress("UNCHECKED_CAST") + (deserialized.actionResults as MutableMap) + .put("__mutable_check__", randomActionRunResult()) + @Suppress("UNCHECKED_CAST") + (deserialized.actionResults as MutableMap) + .remove("__mutable_check__") + } + + // ------------------------------------------------------------------------- + // ChainedAlertTriggerRunResult.actionResults + // ------------------------------------------------------------------------- + + @Test + fun `ChainedAlertTriggerRunResult with empty actionResults survives serialization round-trip`() { + val result = ChainedAlertTriggerRunResult( + triggerName = "chained-trigger", + triggered = false, + error = null, + actionResults = mutableMapOf(), + associatedAlertIds = emptySet() + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = ChainedAlertTriggerRunResult(sin) + + assertEquals(result.triggerName, deserialized.triggerName) + assertTrue(deserialized.actionResults.isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException or ClassCastException + @Suppress("UNCHECKED_CAST") + (deserialized.actionResults as MutableMap) + .put("__mutable_check__", randomActionRunResult()) + @Suppress("UNCHECKED_CAST") + (deserialized.actionResults as MutableMap) + .remove("__mutable_check__") + } + + // ------------------------------------------------------------------------- + // BucketLevelTriggerRunResult.actionResultsMap + // ------------------------------------------------------------------------- + + @Test + fun `BucketLevelTriggerRunResult with empty actionResultsMap survives serialization round-trip`() { + val aggBucket = AggregationResultBucket( + parentBucketPath = "parent_path", + bucketKeys = listOf("key1"), + bucket = mapOf("k" to "v") + ) + val result = BucketLevelTriggerRunResult( + triggerName = "bucket-trigger", + error = null, + aggregationResultBuckets = mapOf(aggBucket.getBucketKeysHash() to aggBucket), + actionResultsMap = mutableMapOf() + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = BucketLevelTriggerRunResult(sin) + + assertEquals(result.triggerName, deserialized.triggerName) + assertTrue(deserialized.actionResultsMap.isEmpty()) + // Verify mutability — must not throw UnsupportedOperationException or ClassCastException + @Suppress("UNCHECKED_CAST") + (deserialized.actionResultsMap as MutableMap>) + .put("__mutable_check__", mutableMapOf()) + @Suppress("UNCHECKED_CAST") + (deserialized.actionResultsMap as MutableMap>) + .remove("__mutable_check__") + } + + // ------------------------------------------------------------------------- + // ActionRunResult.output + // ------------------------------------------------------------------------- + + @Test + fun `ActionRunResult with empty output survives serialization round-trip`() { + val result = ActionRunResult( + actionId = "action-1", + actionName = "test-action", + output = emptyMap(), + throttled = false, + executionTime = Instant.now().truncatedTo(ChronoUnit.MILLIS), + error = null + ) + + val out = BytesStreamOutput() + result.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = ActionRunResult(sin) + + assertEquals(result.actionId, deserialized.actionId) + assertEquals(result.actionName, deserialized.actionName) + assertTrue(deserialized.output.isEmpty()) + // Verify no ClassCastException — output field is Map, backed by mutable map + } + + // ------------------------------------------------------------------------- + // Helper + // ------------------------------------------------------------------------- + + /** + * Asserts that a map is mutable by successfully inserting a sentinel value. + * If the map is Collections.emptyMap() this throws UnsupportedOperationException. + */ + @Suppress("UNCHECKED_CAST") + private fun assertMutableMap(map: Map<*, *>) { + (map as MutableMap)["__mutable_check__"] = "ok" + map.remove("__mutable_check__") + } + + // ------------------------------------------------------------------------- + // BucketSelectorExtAggregationBuilder.bucketsPathsMap + // ------------------------------------------------------------------------- + + @Test + fun `BucketSelectorExtAggregationBuilder with empty bucketsPathsMap survives serialization round-trip`() { + val original = randomBucketSelectorExtAggregationBuilder(bucketsPathsMap = mutableMapOf()) + val out = BytesStreamOutput() + original.writeTo(out) + val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) + val deserialized = BucketSelectorExtAggregationBuilder(sin) + + assertEquals(original.name, deserialized.name) + assertTrue(deserialized.bucketsPathsMap.isEmpty()) + // Verify mutability + assertMutableMap(deserialized.bucketsPathsMap) + } +}