Cherry-pick #21727 to 7.10: Add json body check for sqs message (#22943)

* Add json body check for sqs message (#21727)


(cherry picked from commit 86fc865)

* update changelog

CHANGELOG.next.asciidoc

@@ -162,6 +162,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix `cisco` asa and ftd parsing of messages 106102 and 106103. {pull}20469[20469]
 - Fix event.kind for system/syslog pipeline {issue}20365[20365] {pull}20390[20390]
 - Fix event.type for zeek/ssl and duplicate event.category for zeek/connection {pull}20696[20696]
+- Add json body check for sqs message. {pull}21727[21727]
 - Fix cisco umbrella module config by adding input variable. {pull}22892[22892]
 
 *Heartbeat*

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

@@ -244,10 +244,15 @@ func getRegionFromQueueURL(queueURL string) (string, error) {
 
 // handle message
 func (c *s3Collector) handleSQSMessage(m sqs.Message) ([]s3Info, error) {
-	msg := sqsMessage{}
+	var msg sqsMessage
 	err := json.Unmarshal([]byte(*m.Body), &msg)
 	if err != nil {
-		return nil, fmt.Errorf("json unmarshal sqs message body failed: %w", err)
+		c.logger.Debug("sqs message body = ", *m.Body)
+		if jsonError, ok := err.(*json.SyntaxError); ok {
+			return nil, fmt.Errorf("json unmarshal sqs message body failed at offset %d with syntax error: %w", jsonError.Offset, err)
+		} else {
+			return nil, fmt.Errorf("json unmarshal sqs message body failed: %w", err)
+		}
 	}
 
 	var s3Infos []s3Info