test: add scenario for OTEL context forwarding in parallel bulk

noxfile.py

@@ -52,7 +52,7 @@ def test(session):
     session.run(*pytest_argv())
 
 
-@nox.session(python=["3.7", "3.12"])
+@nox.session(python=["3.7", "3.8", "3.12"])
 def test_otel(session):
     session.install(
         ".[dev]",

test_elasticsearch/test_otel.py

@@ -16,7 +16,9 @@
 #  under the License.
 
 import os
-
+from unittest import mock
+from elasticsearch import Elasticsearch
+from elasticsearch import helpers
 import pytest
 
 try:
@@ -95,3 +97,22 @@ def test_detailed_span():
         "db.elasticsearch.cluster.name": "e9106fc68e3044f0b1475b04bf4ffd5f",
         "db.elasticsearch.node.name": "instance-0000000001",
     }
+
+
+@mock.patch("elasticsearch._otel.OpenTelemetry.recover_parent_context")
+@mock.patch("elasticsearch.helpers.actions._process_bulk_chunk_success")
+@mock.patch("elasticsearch.Elasticsearch.bulk")
+def test_forward_otel_context_to_subthreads(
+    _call_bulk_mock, _process_bulk_success_mock, _mock_otel_recv_context
+):
+    tracer, memory_exporter = setup_tracing()
+    es_client = Elasticsearch("http://localhost:9200")
+    es_client._otel = OpenTelemetry(enabled=True, tracer=tracer)
+
+    _call_bulk_mock.return_value = mock.Mock()
+    actions = ({"x": i} for i in range(100))
+    list(
+        helpers.parallel_bulk(es_client, actions, chunk_size=4)
+    )
+    # Ensures that the OTEL context has been forwarded to all chunks
+    assert es_client._otel.recover_parent_context.call_count == 25