fix chain call issue in HTTP JSON input

CHANGELOG.next.asciidoc

@@ -117,6 +117,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix the ignore_inactive option being ignored in Filebeat's filestream input {pull}34770[34770]
 - Fix TestMultiEventForEOFRetryHandlerInput unit test of CometD input {pull}34903[34903]
 - Add input instance id to request trace filename for httpjson and cel inputs {pull}35024[35024]
+- Fix `httpjson` input storing all the responses in memory for one iteration of chain call. {issue}34310[34310] {pull}34325[34325]
 - Fix panic in TCP and UDP inputs on Linux when collecting socket metrics from OS. {issue}35064[35064]
 - Correctly collect TCP and UDP metrics for unspecified address values. {pull}35111[35111]
 - Fix base for UDP and TCP queue metrics and UDP drops metric. {pull}35123[35123]

x-pack/filebeat/input/httpjson/input_test.go

@@ -13,6 +13,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -35,6 +36,7 @@ var testCases = []struct {
 	expectedFile string
 
 	skipReason string
+	isLongTest bool
 }{
 	{
 		name:        "simple_GET_request",
@@ -728,6 +730,31 @@ var testCases = []struct {
 			`{"space":{"cake":"pumpkin"}}`,
 		},
 	},
+	{
+		name:        "pagination_with_end_condition",
+		setupServer: newPaginationTestServer(httptest.NewServer),
+		baseConfig: map[string]interface{}{
+			"interval":       1,
+			"request.method": http.MethodGet,
+			"response.split": map[string]interface{}{
+				"target": "body.records",
+			},
+			"response.pagination": []interface{}{
+				map[string]interface{}{
+					"set": map[string]interface{}{
+						"target":                 "url.value",
+						"value":                  "[[if (ne .last_response.body.done true)]][[.last_response.body.nextLink]][[end]]",
+						"fail_on_template_error": true,
+					},
+				},
+			},
+		},
+		handler: defaultHandler(http.MethodGet, "", ""),
+		expected: []string{
+			`{"id":1}`,
+			`{"id":2}`,
+		},
+	},
 	{
 		name:        "pagination_when_used_with_chaining",
 		setupServer: newChainPaginationTestServer(httptest.NewServer),
@@ -758,6 +785,68 @@ var testCases = []struct {
 			`{"space":{"cake":"pumpkin"}}`,
 		},
 	},
+	{
+		name:        "pagination_when_used_with_chaining_Unable_to_get_id_from_response",
+		setupServer: newChainPaginationTestServerWithInvalidJSON(httptest.NewServer),
+		baseConfig: map[string]interface{}{
+			"interval":       1,
+			"request.method": http.MethodGet,
+			"response.pagination": []interface{}{
+				map[string]interface{}{
+					"set": map[string]interface{}{
+						"target":                 "url.value",
+						"value":                  "[[.last_response.body.nextLink]]",
+						"fail_on_template_error": true,
+					},
+				},
+			},
+			"chain": []interface{}{
+				map[string]interface{}{
+					"step": map[string]interface{}{
+						"request.method": http.MethodGet,
+						"replace":        "$.records[:].id",
+					},
+				},
+			},
+		},
+		handler:  defaultHandler(http.MethodGet, "", ""),
+		expected: []string{},
+	},
+	{
+		name:        "pagination_when_used_with_chaining_with_bulk_data",
+		setupServer: newChainPaginationLoadTestServer(httptest.NewServer),
+		baseConfig: map[string]interface{}{
+			"interval":       1,
+			"request.method": http.MethodGet,
+			"response.pagination": []interface{}{
+				map[string]interface{}{
+					"set": map[string]interface{}{
+						"target":                 "url.value",
+						"value":                  "[[.last_response.body.nextLink]]",
+						"fail_on_template_error": true,
+					},
+				},
+			},
+			"chain": []interface{}{
+				map[string]interface{}{
+					"step": map[string]interface{}{
+						"request.method": http.MethodGet,
+						"replace":        "$.records[:].id",
+					},
+				},
+			},
+		},
+		handler:    defaultHandler(http.MethodGet, "", ""),
+		isLongTest: true,
+		expected: func() []string {
+			bulkData := make([]string, 5000)
+			text := strings.Repeat(dummyText, 17)
+			for i := range bulkData {
+				bulkData[i] = fmt.Sprintf(`{ "text%d":%q }`, (i/2500)+1, text)
+			}
+			return bulkData
+		}(),
+	},
 	{
 		name: "replace_with_clause_and_first_response_object",
 		setupServer: func(t testing.TB, h http.HandlerFunc, config map[string]interface{}) {
@@ -1181,7 +1270,7 @@ var testCases = []struct {
 		},
 	},
 	{
-		name: "Test simple XML decode",
+		name: "simple_XML_decode",
 		setupServer: func(t testing.TB, h http.HandlerFunc, config map[string]interface{}) {
 			registerDecoders()
 			registerRequestTransforms()
@@ -1205,9 +1294,9 @@ var testCases = []struct {
   </item>
 </order>
 `
-				io.ReadAll(r.Body)
+				io.ReadAll(r.Body) //nolint:errcheck // not expecting any error here
 				r.Body.Close()
-				w.Write([]byte(text))
+				w.Write([]byte(text)) //nolint:errcheck  // not expecting any error here
 			})
 			server := httptest.NewServer(r)
 			t.Cleanup(func() { registeredTransforms = newRegistry() })
@@ -1280,6 +1369,8 @@ var testCases = []struct {
 	},
 }
 
+const dummyText = "Lorem Ipsum is simply dummy text of the printing and typesetting industry. Lorem Ipsum has been the industry's standard dummy text ever since the 1500s, when an unknown printer took a galley of type and scrambled it to make a type specimen book. It has survived not only five centuries, but also the leap into electronic typesetting, remaining essentially unchanged. It was popularised in the 1960s with the release of Letraset sheets containing Lorem Ipsum passages, and more recently with desktop publishing software like Aldus PageMaker including versions of Lorem Ipsum."
+
 func TestInput(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
@@ -1315,8 +1406,11 @@ func TestInput(t *testing.T) {
 			g.Go(func() error {
 				return input.Run(ctx, chanClient)
 			})
-
-			timeout := time.NewTimer(5 * time.Second)
+			timeoutDuration := 10 * time.Second
+			if test.isLongTest {
+				timeoutDuration = 2 * time.Minute
+			}
+			timeout := time.NewTimer(timeoutDuration)
 			t.Cleanup(func() { _ = timeout.Stop() })
 
 			if len(test.expected) == 0 {
@@ -1478,6 +1572,87 @@ func newChainPaginationTestServer(
 	}
 }
 
+func newChainPaginationTestServerWithInvalidJSON(
+	newServer func(http.Handler) *httptest.Server,
+) func(testing.TB, http.HandlerFunc, map[string]interface{}) {
+	return func(t testing.TB, h http.HandlerFunc, config map[string]interface{}) {
+		registerPaginationTransforms()
+		var serverURL string
+		r := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch r.URL.Path {
+			case "/":
+				link := serverURL + "/link2"
+				value := fmt.Sprintf(`"records":[{"id":1}], "nextLink":"%s"}`, link)
+				fmt.Fprintln(w, value)
+			case "/1":
+				fmt.Fprintln(w, `{"hello":{"world":"moon"}}`)
+			}
+		})
+		server := httptest.NewServer(r)
+		config["request.url"] = server.URL
+		serverURL = server.URL
+		config["chain.0.step.request.url"] = server.URL + "/$.records[:].id"
+		t.Cleanup(func() { registeredTransforms = newRegistry() })
+	}
+}
+
+func newPaginationTestServer(
+	newServer func(http.Handler) *httptest.Server,
+) func(testing.TB, http.HandlerFunc, map[string]interface{}) {
+	return func(t testing.TB, h http.HandlerFunc, config map[string]interface{}) {
+		registerPaginationTransforms()
+		registerResponseTransforms()
+		var serverURL string
+		r := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch r.URL.Path {
+			case "/":
+				link := serverURL + "/link2"
+				value := fmt.Sprintf(`{"records":[{"id":1}], "nextLink":"%s", "done": false}`, link)
+				fmt.Fprintln(w, value)
+			case "/link2":
+				link := serverURL + "/"
+				value := fmt.Sprintf(`{"records":[{"id":2}], "nextLink":"%s","done": true}`, link)
+				fmt.Fprintln(w, value)
+			}
+		})
+		server := httptest.NewServer(r)
+		config["request.url"] = server.URL
+		serverURL = server.URL
+		t.Cleanup(func() { registeredTransforms = newRegistry() })
+	}
+}
+
+func newChainPaginationLoadTestServer(
+	newServer func(http.Handler) *httptest.Server,
+) func(testing.TB, http.HandlerFunc, map[string]interface{}) {
+	records1 := strings.Repeat(`{"id":1},`, 2499) + `{"id":1}`
+	records2 := strings.Repeat(`{"id":2},`, 2499) + `{"id":2}`
+
+	return func(t testing.TB, h http.HandlerFunc, config map[string]interface{}) {
+		registerPaginationTransforms()
+		var serverURL string
+		r := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch r.URL.Path {
+			case "/":
+				link := serverURL + "/link2"
+				value := fmt.Sprintf(`{"records":[`+records1+`], "nextLink":"%s"}`, link)
+				fmt.Fprintln(w, value)
+			case "/1":
+				fmt.Fprintf(w, `{ "text1":%q }`+"\n", strings.Repeat(dummyText, 17))
+			case "/link2":
+				fmt.Fprintln(w, `{"records":[`+records2+`]}`)
+			case "/2":
+				fmt.Fprintf(w, `{ "text2":%q }`+"\n", strings.Repeat(dummyText, 17))
+			}
+		})
+		server := httptest.NewServer(r)
+		config["request.url"] = server.URL
+		serverURL = server.URL
+		config["chain.0.step.request.url"] = server.URL + "/$.records[:].id"
+		t.Cleanup(func() { registeredTransforms = newRegistry() })
+	}
+}
+
 func newV2Context(id string) (v2.Context, func()) {
 	ctx, cancel := context.WithCancel(context.Background())
 	return v2.Context{