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

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

CHANGELOG.next.asciidoc

@@ -106,6 +106,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Upgrade azure-event-hubs-go and azure-storage-blob-go dependencies. {pull}38861[38861]
 - Fix request trace filename handling in http_endpoint input. {pull}39410[39410]
 - Upgrade github.com/hashicorp/go-retryablehttp to mitigate CVE-2024-6104 {pull}40036[40036]
+- 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