chore(DSM): fix test flake for dsm processor shutdown (#15821)

## Description

This is just a smaller version of
#15758, so that we can merge
the root fix now without interfering with any changes that would come
from this PR: #15715.

Tests were flaky due to:
Race during teardown where processor.shutdown() could be called twice
(once by the test fixture and once by global tracer teardown), causing a
ServiceStatusError; the new try/except ignores that expected case.

## Testing

<!-- Describe your testing strategy or note what tests are included -->

## Risks

<!-- Note any risks associated with this change, or "None" if no risks
-->

## Additional Notes

<!-- Any other information that would be helpful for reviewers -->

Author: robcarlan-datadog

tests/contrib/aiokafka/test_aiokafka_dsm.py

@@ -9,6 +9,8 @@
 from ddtrace.internal.datastreams.processor import DataStreamsCtx
 from ddtrace.internal.datastreams.processor import PartitionKey
 from ddtrace.internal.native import DDSketch
+from ddtrace.internal.service import ServiceStatus
+from ddtrace.internal.service import ServiceStatusError
 from tests.utils import DummyTracer
 from tests.utils import override_global_tracer
 
@@ -48,7 +50,13 @@ def dsm_processor(tracer):
 
     with mock.patch("ddtrace.internal.datastreams.data_streams_processor", return_value=processor):
         yield processor
-        processor.shutdown(timeout=5)
+
+        try:
+            processor.shutdown(timeout=5)
+        except ServiceStatusError as e:
+            # Expected: processor already stopped by tracer shutdown during test teardown
+            if e.current_status == ServiceStatus.RUNNING:
+                raise
 
 
 @pytest.mark.asyncio

tests/contrib/kafka/test_kafka_dsm.py

@@ -11,6 +11,8 @@
 from ddtrace.internal.datastreams.processor import DataStreamsCtx
 from ddtrace.internal.datastreams.processor import PartitionKey
 from ddtrace.internal.native import DDSketch
+from ddtrace.internal.service import ServiceStatus
+from ddtrace.internal.service import ServiceStatusError
 from tests.datastreams.test_public_api import MockedTracer
 
 
@@ -24,11 +26,24 @@ class CustomError(Exception):
 @pytest.fixture
 def dsm_processor(tracer):
     processor = tracer.data_streams_processor
+    # Clean up any existing context to prevent test pollution
+    try:
+        del processor._current_context.value
+    except AttributeError:
+        pass
+
     with mock.patch("ddtrace.internal.datastreams.data_streams_processor", return_value=processor):
         yield processor
         # flush buckets for the next test run
         processor.periodic()
 
+        try:
+            processor.shutdown(timeout=5)
+        except ServiceStatusError as e:
+            # Expected: processor already stopped by tracer shutdown during test teardown
+            if e.current_status == ServiceStatus.RUNNING:
+                raise
+
 
 @pytest.mark.parametrize("payload_and_length", [("test", 4), ("你".encode("utf-8"), 3), (b"test2", 5)])
 @pytest.mark.parametrize("key_and_length", [("test-key", 8), ("你".encode("utf-8"), 3), (b"t2", 2)])
@@ -44,11 +59,6 @@ def test_data_streams_payload_size(dsm_processor, consumer, producer, kafka_topi
     expected_payload_size += len(PROPAGATION_KEY_BASE_64)  # Add in header key length
     expected_payload_size += DSM_TEST_PATH_HEADER_SIZE  # to account for path header we add
 
-    try:
-        del dsm_processor._current_context.value
-    except AttributeError:
-        pass
-
     producer.produce(kafka_topic, payload, key=key, headers=test_headers)
     producer.flush()
     consumer.poll()
@@ -65,10 +75,6 @@ def test_data_streams_payload_size(dsm_processor, consumer, producer, kafka_topi
 
 def test_data_streams_kafka_serializing(dsm_processor, deserializing_consumer, serializing_producer, kafka_topic):
     PAYLOAD = bytes("data streams", encoding="utf-8")
-    try:
-        del dsm_processor._current_context.value
-    except AttributeError:
-        pass
     serializing_producer.produce(kafka_topic, value=PAYLOAD, key="test_key_2")
     serializing_producer.flush()
     message = None
@@ -80,10 +86,6 @@ def test_data_streams_kafka_serializing(dsm_processor, deserializing_consumer, s
 
 def test_data_streams_kafka(dsm_processor, consumer, producer, kafka_topic):
     PAYLOAD = bytes("data streams", encoding="utf-8")
-    try:
-        del dsm_processor._current_context.value
-    except AttributeError:
-        pass
     producer.produce(kafka_topic, PAYLOAD, key="test_key_1")
     producer.produce(kafka_topic, PAYLOAD, key="test_key_2")
     producer.flush()
@@ -127,10 +129,6 @@ def _read_single_message(consumer):
 
     PAYLOAD = bytes("data streams", encoding="utf-8")
     consumer = non_auto_commit_consumer
-    try:
-        del dsm_processor._current_context.value
-    except AttributeError:
-        pass
     buckets = dsm_processor._buckets
     producer.produce(kafka_topic, PAYLOAD, key="test_key_1")
     producer.produce(kafka_topic, PAYLOAD, key="test_key_2")
@@ -170,10 +168,6 @@ def _read_single_message(consumer):
 
     consumer = non_auto_commit_consumer
     PAYLOAD = bytes("data streams", encoding="utf-8")
-    try:
-        del dsm_processor._current_context.value
-    except AttributeError:
-        pass
     producer.produce(kafka_topic, PAYLOAD, key="test_key_1")
     producer.produce(kafka_topic, PAYLOAD, key="test_key_2")
     producer.flush()
@@ -207,10 +201,6 @@ def _read_single_message(consumer):
                 return message
 
     PAYLOAD = bytes("data streams", encoding="utf-8")
-    try:
-        del dsm_processor._current_context.value
-    except AttributeError:
-        pass
     producer.produce(kafka_topic, PAYLOAD, key="test_key_1")
     producer.produce(kafka_topic, PAYLOAD, key="test_key_2")
     producer.flush()
@@ -236,10 +226,6 @@ def test_data_streams_kafka_produce_api_compatibility(dsm_processor, consumer, p
     kafka_topic = empty_kafka_topic
 
     PAYLOAD = bytes("data streams", encoding="utf-8")
-    try:
-        del dsm_processor._current_context.value
-    except AttributeError:
-        pass
 
     # All of these should work
     producer.produce(kafka_topic)