Add support for array parsing in azure-eventhub input

CHANGELOG.next.asciidoc

@@ -333,6 +333,7 @@ field. You can revert this change by configuring tags for the module and omittin
 - Improve ECS categorization field mappings in system module. {issue}16031[16031] {pull}18065[18065]
 - Change the `json.*` input settings implementation to merge parsed json objects with existing objects in the event instead of fully replacing them. {pull}17958[17958]
 - Improve ECS categorization field mappings in osquery module. {issue}16176[16176] {pull}17881[17881]
+- Add support for array parsing in azure-eventhub input. {pull}18585[18585]
 - Added `observer.vendor`, `observer.product`, and `observer.type` to PANW module events. {pull}18223[18223]
 - The `logstash` module can now automatically detect the log file format (JSON or plaintext) and process it accordingly. {issue}9964[9964] {pull}18095[18095]
 - Improve ECS categorization field mappings in envoyproxy module. {issue}16161[16161] {pull}18395[18395]

x-pack/filebeat/input/azureeventhub/input.go

@@ -203,14 +203,27 @@ func (a *azureInput) processEvents(event *eventhub.Event, partitionID string) bo
 
 // parseMultipleMessages will try to split the message into multiple ones based on the group field provided by the configuration
 func (a *azureInput) parseMultipleMessages(bMessage []byte) []string {
-	var obj map[string][]interface{}
-	err := json.Unmarshal(bMessage, &obj)
-	if err != nil {
-		a.log.Errorw(fmt.Sprintf("deserializing multiple messages using the group object `records`"), "error", err)
-	}
+	var mapObject map[string][]interface{}
 	var messages []string
-	if len(obj[expandEventListFromField]) > 0 {
-		for _, ms := range obj[expandEventListFromField] {
+	// check if the message is an object containing a list of events
+	err := json.Unmarshal(bMessage, &mapObject)
+	if err == nil {
+		if len(mapObject[expandEventListFromField]) > 0 {
+			for _, ms := range mapObject[expandEventListFromField] {
+				js, err := json.Marshal(ms)
+				if err == nil {
+					messages = append(messages, string(js))
+				} else {
+					a.log.Errorw(fmt.Sprintf("serializing message %s", ms), "error", err)
+				}
+			}
+		}
+	} else {
+		// in some cases the message is an array
+		a.log.Debugw(fmt.Sprintf("deserializing multiple messages using the group object `records`"), "warning", err)
+		var arrayObject []interface{}
+		err = json.Unmarshal(bMessage, &arrayObject)
+		for _, ms := range arrayObject {
 			js, err := json.Marshal(ms)
 			if err == nil {
 				messages = append(messages, string(js))