[8.17](backport #42036) [filebeat][streaming] - Fix for streaming inp…

…ut handling of invalid or empty websocket messages (#42049)

* [filebeat][streaming] - Fix for streaming input handling of invalid or empty websocket messages (#42036)

* Fix for streaming input handling of invalid or empty websocket messages

(cherry picked from commit d508a40)

* Update CHANGELOG.next.asciidoc

---------

Co-authored-by: ShourieG <shourie.ganguly@elastic.co>

CHANGELOG.next.asciidoc

@@ -114,6 +114,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix publication of group data from the Okta entity analytics provider. {pull}40681[40681]
 - Ensure netflow custom field configuration is applied. {issue}40735[40735] {pull}40730[40730]
 - Fix a bug in Salesforce input to only handle responses with 200 status code {pull}41015[41015]
+- Fix streaming input handling of invalid or empty websocket messages. {pull}42036[42036]
 
 *Heartbeat*
 

x-pack/filebeat/input/streaming/websocket.go

@@ -118,25 +118,25 @@ func (s *websocketStream) FollowStream(ctx context.Context) error {
 			_, message, err := c.ReadMessage()
 			if err != nil {
 				s.metrics.errorsTotal.Inc()
-				if isRetryableError(err) {
-					s.log.Debugw("websocket connection encountered an error, attempting to reconnect...", "error", err)
-					// close the old connection and reconnect
-					if err := c.Close(); err != nil {
-						s.metrics.errorsTotal.Inc()
-						s.log.Errorw("encountered an error while closing the websocket connection", "error", err)
-					}
-					// since c is already a pointer, we can reassign it to the new connection and the defer func will still handle it
-					c, resp, err = connectWebSocket(ctx, s.cfg, url, s.log)
-					handleConnectionResponse(resp, s.metrics, s.log)
-					if err != nil {
-						s.metrics.errorsTotal.Inc()
-						s.log.Errorw("failed to reconnect websocket connection", "error", err)
-						return err
-					}
-				} else {
+				if !isRetryableError(err) {
 					s.log.Errorw("failed to read websocket data", "error", err)
 					return err
 				}
+				s.log.Debugw("websocket connection encountered an error, attempting to reconnect...", "error", err)
+				// close the old connection and reconnect
+				if err := c.Close(); err != nil {
+					s.metrics.errorsTotal.Inc()
+					s.log.Errorw("encountered an error while closing the websocket connection", "error", err)
+				}
+				// since c is already a pointer, we can reassign it to the new connection and the defer func will still handle it
+				c, resp, err = connectWebSocket(ctx, s.cfg, url, s.log)
+				handleConnectionResponse(resp, s.metrics, s.log)
+				if err != nil {
+					s.metrics.errorsTotal.Inc()
+					s.log.Errorw("failed to reconnect websocket connection", "error", err)
+					return err
+				}
+				continue
 			}
 			s.metrics.receivedBytesTotal.Add(uint64(len(message)))
 			state["response"] = message