chore: Bump sentry-arroyo to 2.18.2 (#81270)

Bumping this to latest for some fixes

<!-- Describe your PR here. -->

---------

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>
Co-authored-by: Markus Unterwaditzer <[email protected]>
Co-authored-by: Josh Ferge <[email protected]>

Author: 4 people

requirements-base.txt

@@ -65,7 +65,7 @@ requests>=2.32.3
 rfc3339-validator>=0.1.2
 rfc3986-validator>=0.1.1
 # [end] jsonschema format validators
-sentry-arroyo>=2.16.5
+sentry-arroyo>=2.18.2
 sentry-kafka-schemas>=0.1.122
 sentry-ophio==1.0.0
 sentry-protos>=0.1.37

requirements-dev-frozen.txt

@@ -180,7 +180,7 @@ rpds-py==0.20.0
 rsa==4.8
 s3transfer==0.10.0
 selenium==4.16.0
-sentry-arroyo==2.16.5
+sentry-arroyo==2.18.2
 sentry-cli==2.16.0
 sentry-covdefaults-disable-branch-coverage==1.0.2
 sentry-devenv==1.14.2

requirements-frozen.txt

@@ -124,7 +124,7 @@ rfc3986-validator==0.1.1
 rpds-py==0.20.0
 rsa==4.8
 s3transfer==0.10.0
-sentry-arroyo==2.16.5
+sentry-arroyo==2.18.2
 sentry-kafka-schemas==0.1.122
 sentry-ophio==1.0.0
 sentry-protos==0.1.37

src/sentry/consumers/synchronized.py

@@ -300,3 +300,7 @@ def close(self, timeout: float | None = None) -> None:
     @property
     def closed(self) -> bool:
         return self.__consumer.closed
+
+    @property
+    def member_id(self) -> str:
+        return self.__consumer.member_id

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

@@ -2,6 +2,7 @@
 import logging
 from collections import defaultdict
 from collections.abc import Mapping
+from datetime import datetime
 from typing import Any
 
 import orjson
@@ -15,7 +16,15 @@
 from arroyo.processing.strategies.produce import Produce
 from arroyo.processing.strategies.run_task import RunTask
 from arroyo.processing.strategies.unfold import Unfold
-from arroyo.types import FILTERED_PAYLOAD, BrokerValue, Commit, FilteredPayload, Message, Partition
+from arroyo.types import (
+    FILTERED_PAYLOAD,
+    BrokerValue,
+    Commit,
+    FilteredPayload,
+    Message,
+    Partition,
+    Value,
+)
 from sentry_kafka_schemas.codecs import Codec
 from sentry_kafka_schemas.schema_types.snuba_spans_v1 import SpanEvent
 
@@ -196,7 +205,7 @@ def batch_write_to_redis(
 
 def _expand_segments(should_process_segments: list[ProcessSegmentsContext]):
     with sentry_sdk.start_transaction(op="process", name="spans.process.expand_segments") as txn:
-        buffered_segments: list[KafkaPayload | FilteredPayload] = []
+        buffered_segments: list[Value] = []
 
         for result in should_process_segments:
             timestamp = result.timestamp
@@ -235,7 +244,13 @@ def _expand_segments(should_process_segments: list[ProcessSegmentsContext]):
                             metrics.incr("performance.buffered_segments.max_payload_size_exceeded")
                             continue
 
-                        buffered_segments.append(KafkaPayload(None, payload_data, []))
+                        buffered_segments.append(
+                            Value(
+                                KafkaPayload(None, payload_data, []),
+                                {},
+                                datetime.fromtimestamp(timestamp),
+                            )
+                        )
 
     return buffered_segments