Merge of #21214

Author: mergify[bot]

systemtest/estest/search.go

@@ -77,7 +77,15 @@ func ExpectMinDocs(t testing.TB, es *espoll.Client, min int, index string, query
 	return result
 }
 
-func ExpectSourcemapError(t testing.TB, es *espoll.Client, index string, retry func(), query interface{}, updated bool) espoll.SearchResult {
+// ExpectSourcemapError retries sending events via retry until the sourcemap
+// fetcher is available (i.e. no "fetcher unavailable" errors appear in the
+// indexed documents), then returns the search result.
+//
+// minDocs is the minimum number of documents expected in the index after a
+// successful retry. Callers must pass the exact count they expect so that the
+// underlying poll does not return early when Elasticsearch has only indexed a
+// partial batch (see: TotalHitsCondition race).
+func ExpectSourcemapError(t testing.TB, es *espoll.Client, index string, minDocs int, retry func(), query interface{}, updated bool) espoll.SearchResult {
 	t.Helper()
 
 	deadline := time.After(5 * time.Second)
@@ -99,7 +107,7 @@ func ExpectSourcemapError(t testing.TB, es *espoll.Client, index string, retry f
 
 			retry()
 
-			result := ExpectDocs(t, es, index, query)
+			result := ExpectMinDocs(t, es, minDocs, index, query)
 
 			if isFetcherAvailable(t, result) {
 				assertSourcemapUpdated(t, result, updated)

systemtest/rum_test.go

@@ -220,7 +220,11 @@ func TestRUMRoutingIntegration(t *testing.T) {
 		require.NoError(t, err)
 		defer resp.Body.Close()
 	}
-	result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "traces-apm.rum*", retry, nil, false)
+	// 9 = 1 transaction + 8 spans produced by testdata/intake-v3/rum_events.ndjson.
+	// Passing the exact count prevents ExpectSourcemapError from returning early
+	// when Elasticsearch has only indexed a partial batch (TotalHitsCondition race).
+	const wantDocs = 9
+	result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "traces-apm.rum*", wantDocs, retry, nil, false)
 	approvaltest.ApproveFields(
 		t, t.Name(), result.Hits.Hits, "@timestamp", "timestamp.us",
 		"source.port", "source.ip", "client.ip",

systemtest/sourcemap_test.go

@@ -43,7 +43,7 @@ func TestRUMErrorSourcemapping(t *testing.T) {
 			retry := func() {
 				systemtest.SendRUMEventsPayload(t, serverURL, "../testdata/intake-v2/errors_rum.ndjson")
 			}
-			result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", retry, nil, true)
+			result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", 1, retry, nil, true)
 			approvaltest.ApproveFields(
 				t, t.Name(), result.Hits.Hits,
 				// RUM timestamps are set by the server based on the time the payload is received.
@@ -92,7 +92,7 @@ func TestRUMSpanSourcemapping(t *testing.T) {
 	retry := func() {
 		systemtest.SendRUMEventsPayload(t, srv.URL, "../testdata/intake-v2/transactions_spans_rum_2.ndjson")
 	}
-	result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "traces-apm*", retry, espoll.TermQuery{
+	result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "traces-apm*", 1, retry, espoll.TermQuery{
 		Field: "processor.event",
 		Value: "span",
 	}, true)
@@ -125,7 +125,7 @@ func TestNoMatchingSourcemap(t *testing.T) {
 	retry := func() {
 		systemtest.SendRUMEventsPayload(t, srv.URL, "../testdata/intake-v2/transactions_spans_rum_2.ndjson")
 	}
-	result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "traces-apm*", retry, espoll.TermQuery{
+	result := estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "traces-apm*", 1, retry, espoll.TermQuery{
 		Field: "processor.event",
 		Value: "span",
 	}, false)
@@ -158,11 +158,11 @@ func TestSourcemapCaching(t *testing.T) {
 	retry := func() {
 		systemtest.SendRUMEventsPayload(t, srv.URL, "../testdata/intake-v2/errors_rum.ndjson")
 	}
-	estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", retry, nil, true)
+	estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", 1, retry, nil, true)
 
 	// Delete the source map and error, and try again.
 	systemtest.CleanupElasticsearch(t)
-	estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", retry, nil, true)
+	estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", 1, retry, nil, true)
 }
 
 func TestSourcemapFetcher(t *testing.T) {
@@ -217,7 +217,7 @@ func TestSourcemapFetcher(t *testing.T) {
 			retry := func() {
 				systemtest.SendRUMEventsPayload(t, srv.URL, "../testdata/intake-v2/errors_rum.ndjson")
 			}
-			estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", retry, nil, true)
+			estest.ExpectSourcemapError(t, systemtest.Elasticsearch, "logs-apm.error-*", 1, retry, nil, true)
 		})
 	}
 }