[chore][exporter/elasticsearch] Fix flaky integration test by 1:1 bulk request to Consume correspondence (open-telemetry#41276)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Fix flaky integration test with received > sent, caused by race
condition where Consume* returns an error halfway through processing a
bulk request, causing duplicates in the mock backend because the first N
documents went through and an emulated http error causes the entire
request to be retried, including the first N documents.

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#41208
Fixes open-telemetry#41259

<!--Describe what testing was performed and which tests were added.-->
#### Testing

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

Author: carsonip

exporter/elasticsearchexporter/integrationtest/datareceiver.go

@@ -275,17 +275,6 @@ func (es *mockESReceiver) Start(ctx context.Context, host component.Host) error
 		return fmt.Errorf("failed to bind to address %s: %w", es.config.Endpoint, err)
 	}
 
-	// Ideally bulk request items should be converted to the corresponding event record
-	// however, since we only assert count for now there is no need to do the actual
-	// translation. Instead we use a pre-initialized empty models to
-	// reduce allocation impact on tests and benchmarks.
-	emptyLogs := plog.NewLogs()
-	emptyLogs.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
-	emptyMetrics := pmetric.NewMetrics()
-	emptyMetrics.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetEmptySum().DataPoints().AppendEmpty()
-	emptyTrace := ptrace.NewTraces()
-	emptyTrace.ResourceSpans().AppendEmpty().ScopeSpans().AppendEmpty().Spans().AppendEmpty()
-
 	r := mux.NewRouter()
 	r.Use(func(next http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -301,29 +290,66 @@ func (es *mockESReceiver) Start(ctx context.Context, host component.Host) error
 			fmt.Fprintln(w, "{}")
 			return
 		}
+		var index string
+		var itemCount int
 		_, response := docappendertest.DecodeBulkRequest(r)
 		for _, itemMap := range response.Items {
-			for k, item := range itemMap {
-				var consumeErr error
-				switch item.Index {
-				case TestLogsIndex:
-					consumeErr = es.logsConsumer.ConsumeLogs(context.Background(), emptyLogs)
-				case TestMetricsIndex:
-					consumeErr = es.metricsConsumer.ConsumeMetrics(context.Background(), emptyMetrics)
-				case TestTracesIndex:
-					consumeErr = es.tracesConsumer.ConsumeTraces(context.Background(), emptyTrace)
+			for _, item := range itemMap {
+				if index == "" {
+					index = item.Index
+				} else if item.Index != index {
+					panic("mock ES receiver assumes that all documents target the same index")
 				}
-				var errES errElasticsearch
-				if consumeErr != nil {
-					if !errors.As(consumeErr, &errES) {
-						// panic to surface test logic error because we only expect error of type errElasticsearch
-						panic("unknown consume error")
-					}
-					if errES.httpStatus != http.StatusOK {
-						w.WriteHeader(errES.httpStatus)
-						return
-					}
-					response.HasErrors = true
+				itemCount++
+			}
+		}
+
+		// Assuming all documents are of the same type (logs, metrics, traces),
+		// create a pdata struct with the same number of records and send them in 1 Consume* call,
+		// i.e. a 1:1 bulk request to Consume* function call correspondence.
+		// This avoids a race condition where Consume* returns an error halfway through processing a bulk request,
+		// causing duplicates in the mock backend because the first N documents went through and an emulated http error
+		// causes the entire request to be retried, including the first N documents.
+		var consumeErr error
+		switch index {
+		case TestLogsIndex:
+			emptyLogs := plog.NewLogs()
+			lr := emptyLogs.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords()
+			for range itemCount {
+				lr.AppendEmpty()
+			}
+			emptyLogs.MarkReadOnly()
+			consumeErr = es.logsConsumer.ConsumeLogs(context.Background(), emptyLogs)
+		case TestMetricsIndex:
+			emptyMetrics := pmetric.NewMetrics()
+			dp := emptyMetrics.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetEmptySum().DataPoints()
+			for range itemCount {
+				dp.AppendEmpty()
+			}
+			emptyMetrics.MarkReadOnly()
+			consumeErr = es.metricsConsumer.ConsumeMetrics(context.Background(), emptyMetrics)
+		case TestTracesIndex:
+			emptyTrace := ptrace.NewTraces()
+			spans := emptyTrace.ResourceSpans().AppendEmpty().ScopeSpans().AppendEmpty().Spans()
+			for range itemCount {
+				spans.AppendEmpty()
+			}
+			emptyTrace.MarkReadOnly()
+			consumeErr = es.tracesConsumer.ConsumeTraces(context.Background(), emptyTrace)
+		}
+		if consumeErr != nil {
+			var errES errElasticsearch
+			if !errors.As(consumeErr, &errES) {
+				// panic to surface test logic error because we only expect error of type errElasticsearch
+				panic("unknown consume error")
+			}
+			if errES.httpStatus != http.StatusOK {
+				w.WriteHeader(errES.httpStatus)
+				return
+			}
+			response.HasErrors = true
+			for _, itemMap := range response.Items {
+				for k, item := range itemMap {
 					item.Status = errES.httpDocStatus
 					item.Error.Type = "simulated_es_error"
 					item.Error.Reason = consumeErr.Error()