From b62e4de52c143b8880199151dc557058858dfa72 Mon Sep 17 00:00:00 2001
From: subham sarkar <subham.sarkar@elastic.co>
Date: Tue, 4 Jun 2024 01:23:29 +0530
Subject: [PATCH] x-pack/filebeat/input/httpjson: Close connections properly

---
 x-pack/filebeat/input/httpjson/policy.go  |  5 +++++
 x-pack/filebeat/input/httpjson/request.go | 15 +++++++--------
 2 files changed, 12 insertions(+), 8 deletions(-)

diff --git a/x-pack/filebeat/input/httpjson/policy.go b/x-pack/filebeat/input/httpjson/policy.go
index 0c671cb85bbb..5c140eba6624 100644
--- a/x-pack/filebeat/input/httpjson/policy.go
+++ b/x-pack/filebeat/input/httpjson/policy.go
@@ -91,6 +91,11 @@ func (p *Policy) CustomRetryPolicy(ctx context.Context, resp *http.Response, err
 	// errors and may relate to outages on the server side. This will catch
 	// invalid response codes as well, like 0 and 999.
 	if resp.StatusCode == 0 || (resp.StatusCode >= 500 && resp.StatusCode != 501) {
+		defer func() {
+			if resp.Body != nil {
+				resp.Body.Close()
+			}
+		}()
 		return true, nil
 	}
 
diff --git a/x-pack/filebeat/input/httpjson/request.go b/x-pack/filebeat/input/httpjson/request.go
index 3e63f0267162..6e45487f8b34 100644
--- a/x-pack/filebeat/input/httpjson/request.go
+++ b/x-pack/filebeat/input/httpjson/request.go
@@ -618,7 +618,6 @@ func (r *requester) processChainPaginationEvents(ctx context.Context, trCtx *tra
 		err               error
 		urlCopy           url.URL
 		urlString         string
-		httpResp          *http.Response
 		intermediateResps []*http.Response
 		finalResps        []*http.Response
 	)
@@ -672,10 +671,16 @@ func (r *requester) processChainPaginationEvents(ctx context.Context, trCtx *tra
 			}
 
 			// collect data from new urls
-			httpResp, err = rf.collectResponse(ctx, chainTrCtx, r)
+			httpResp, err := rf.collectResponse(ctx, chainTrCtx, r)
 			if err != nil {
 				return -1, fmt.Errorf("failed to collect response: %w", err)
 			}
+			defer func() {
+				if httpResp != nil && httpResp.Body != nil {
+					httpResp.Body.Close()
+				}
+			}()
+
 			// store data according to response type
 			if i == len(r.requestFactories)-1 && len(ids) != 0 {
 				finalResps = append(finalResps, httpResp)
@@ -702,12 +707,6 @@ func (r *requester) processChainPaginationEvents(ctx context.Context, trCtx *tra
 		n += p.eventCount()
 	}
 
-	defer func() {
-		if httpResp != nil && httpResp.Body != nil {
-			httpResp.Body.Close()
-		}
-	}()
-
 	return n, nil
 }