[Filebeat] return error when expand_event_list_from_field is missing (e…

…lastic#17121)

* return error when expand_event_list_from_field is missing with application/json file

x-pack/filebeat/docs/inputs/input-aws-s3.asciidoc

@@ -64,6 +64,8 @@ are in JSON format and events are found under the JSON object "Records".
 Note: When `expand_event_list_from_field` parameter is given in the config, s3
 input will assume the logs are in JSON format and decode them as JSON. Content
 type will not be checked.
+If a file has "application/json" content-type, `expand_event_list_from_field`
+becomes required to read the json file.
 
 [float]
 ==== `api_timeout`

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

@@ -428,6 +428,13 @@ func (p *s3Input) createEventsFromS3Info(svc s3iface.ClientAPI, info s3Info, s3C
 
 	reader := bufio.NewReader(resp.Body)
 
+	// Check if expand_event_list_from_field is given with document conent-type = "application/json"
+	if resp.ContentType != nil && *resp.ContentType == "application/json" && p.config.ExpandEventListFromField == "" {
+		err := errors.New("expand_event_list_from_field parameter is missing in config for application/json content-type file")
+		p.logger.Error(err)
+		return err
+	}
+
 	// Decode JSON documents when expand_event_list_from_field is given in config
 	if p.config.ExpandEventListFromField != "" {
 		decoder := json.NewDecoder(reader)
@@ -440,7 +447,7 @@ func (p *s3Input) createEventsFromS3Info(svc s3iface.ClientAPI, info s3Info, s3C
 		return nil
 	}
 
-	// Check content-type
+	// Check content-type = "application/x-gzip" or filename ends with ".gz"
 	if (resp.ContentType != nil && *resp.ContentType == "application/x-gzip") || strings.HasSuffix(info.key, ".gz") {
 		gzipReader, err := gzip.NewReader(resp.Body)
 		if err != nil {