test(spans): V1 kafka payload (#57449)

I tried sending in a JSON payload to the spans consumer and it seems to
catch the wrong exception. Changed to `ValueError` to catch both
`FormatError` and `ExtraData`.

src/sentry/spans/consumers/process/factory.py

@@ -168,7 +168,7 @@ def _deserialize_payload(payload: bytes) -> Mapping[str, Any]:
     # This for backward compatibility while we transition.
     try:
         return msgpack.unpackb(payload)
-    except msgpack.FormatError:
+    except ValueError:
         return json.loads(payload, use_rapid_json=True)
 
 

tests/sentry/spans/test_spans_process_consumer.py

@@ -9,9 +9,11 @@ from arroyo.types import BrokerValue, Message, Partition, Topic
 from sentry.receivers import create_default_projects
 from sentry.spans.consumers.process.factory import (
     ProcessSpansStrategyFactory,
+    _process_message,
     _process_relay_span_v0,
 )
 from sentry.testutils.pytest.fixtures import django_db_all
+from sentry.utils import json
 
 
 @django_db_all
@@ -132,3 +134,67 @@ def test_null_tags_and_data():
 
     assert "group" not in snuba_span["sentry_tags"]
     assert "status_code" not in snuba_span["sentry_tags"]
+
+
+def test_v1_span():
+    # Taken from https://github.com/getsentry/relay/blob/e27eff09c32f2604e5b9e4533d891dabdd52978a/tests/integration/test_store.py#L1225-L1261
+    payload = json.dumps(
+        {
+            "event_id": "cbf6960622e14a45abc1f03b2055b186",
+            "project_id": 42,
+            "organization_id": 1,
+            "retention_days": 90,
+            "span": {
+                "description": "GET /api/0/organizations/?member=1",
+                "exclusive_time": 500.0,
+                "is_segment": False,
+                "op": "http",
+                "parent_span_id": "aaaaaaaaaaaaaaaa",
+                "segment_id": "968cff94913ebb07",
+                "sentry_tags": {
+                    "category": "http",
+                    "description": "GET *",
+                    "group": "37e3d9fab1ae9162",
+                    "module": "http",
+                    "op": "http",
+                    "transaction": "hi",
+                    "transaction.op": "hi",
+                },
+                "span_id": "bbbbbbbbbbbbbbbb",
+                "start_timestamp": 123.456,
+                "timestamp": 124.567,
+                "trace_id": "ff62a8b040f340bda5d830223def1d81",
+            },
+        }
+    ).encode()
+    value = BrokerValue(KafkaPayload(None, payload, []), None, 0, None)  # type: ignore
+    processed = _process_message(Message(value))
+    assert json.loads(processed.value) == {
+        "description": "GET /api/0/organizations/?member=1",
+        "duration_ms": 1111,
+        "event_id": "cbf6960622e14a45abc1f03b2055b186",
+        "exclusive_time_ms": 500,
+        "group_raw": "3f9ccdec3e17d794",
+        "is_segment": False,
+        "organization_id": 1,
+        "parent_span_id": "aaaaaaaaaaaaaaaa",
+        "project_id": 42,
+        "retention_days": 90,
+        "segment_id": "968cff94913ebb07",
+        "sentry_tags": {
+            "description": "GET *",
+            "group": "37e3d9fab1ae9162",
+            "module": "http",
+            "op": "http",
+            "category": "http",
+            "status": "",
+            "transaction": "hi",
+            "transaction.op": "hi",
+        },
+        "span_grouping_config": {"id": "default:2022-10-27"},
+        "span_id": "bbbbbbbbbbbbbbbb",
+        "start_timestamp_ms": 123456,
+        "tags": {},
+        "trace_id": "ff62a8b040f340bda5d830223def1d81",
+        "version": 1,
+    }