[filebeat][streaming] - Fix for streaming input handling of invalid o…

…r empty websocket messages (#42036)

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

CHANGELOG.next.asciidoc

@@ -195,6 +195,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Rate limiting fixes in the Okta provider of the Entity Analytics input. {issue}40106[40106] {pull}41583[41583]
 - Redact authorization headers in HTTPJSON debug logs. {pull}41920[41920]
 - Further rate limiting fix in the Okta provider of the Entity Analytics input. {issue}40106[40106] {pull}41977[41977]
+- 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