build: Upgrade arroyo to 2.20 (#87289)

Author: jan-auer

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.19.9
+sentry-arroyo>=2.20.0
 sentry-kafka-schemas>=1.1.2
 sentry-ophio==1.0.0
 sentry-protos>=0.1.62

requirements-dev-frozen.txt

@@ -29,7 +29,7 @@ click==8.1.7
 click-didyoumean==0.3.0
 click-plugins==1.1.1
 click-repl==0.3.0
-confluent-kafka==2.3.0
+confluent-kafka==2.8.0
 covdefaults==2.3.0
 coverage==7.6.4
 cronsim==2.6
@@ -180,7 +180,7 @@ rpds-py==0.20.0
 rsa==4.8
 s3transfer==0.10.0
 selenium==4.16.0
-sentry-arroyo==2.19.9
+sentry-arroyo==2.20.0
 sentry-cli==2.16.0
 sentry-covdefaults-disable-branch-coverage==1.0.2
 sentry-devenv==1.16.0

requirements-frozen.txt

@@ -26,7 +26,7 @@ click==8.1.7
 click-didyoumean==0.3.0
 click-plugins==1.1.1
 click-repl==0.3.0
-confluent-kafka==2.3.0
+confluent-kafka==2.8.0
 cronsim==2.6
 cryptography==43.0.1
 cssselect==1.0.3
@@ -123,7 +123,7 @@ rfc3986-validator==0.1.1
 rpds-py==0.20.0
 rsa==4.8
 s3transfer==0.10.0
-sentry-arroyo==2.19.9
+sentry-arroyo==2.20.0
 sentry-forked-email-reply-parser==0.5.12.post1
 sentry-kafka-schemas==1.1.2
 sentry-ophio==1.0.0

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

@@ -9,7 +9,6 @@
 from arroyo.processing.strategies.abstract import ProcessingStrategy, ProcessingStrategyFactory
 from arroyo.processing.strategies.commit import CommitOffsets
 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 Commit, FilteredPayload, Message, Partition, Value
 from sentry_kafka_schemas.codecs import Codec
@@ -67,17 +66,11 @@ def create_with_partitions(
         else:
             produce_step = commit_step
 
-        # XXX: Remove after https://github.com/getsentry/arroyo/pull/427: Unfold
-        # does not pass through the commit and there is no way to access it from
-        # the generator function.
-        zip_commit = RunTask(
-            function=lambda m: (m.payload, m.committable),
-            next_step=Unfold(generator=_unfold_segment, next_step=produce_step),
-        )
+        unfold_step = Unfold(generator=_unfold_segment, next_step=produce_step)
 
         return run_task_with_multiprocessing(
             function=_process_message,
-            next_step=zip_commit,
+            next_step=unfold_step,
             max_batch_size=self.max_batch_size,
             max_batch_time=self.max_batch_time,
             pool=self.pool,
@@ -105,13 +98,11 @@ def _process_message(message: Message[KafkaPayload]) -> list[Span]:
         # raise InvalidMessage(message.value.partition, message.value.offset)
 
 
-def _unfold_segment(message: tuple[list[Span], Mapping[Partition, int]]):
-    spans, committable = message
-    last = len(spans) - 1
-    for i, span in enumerate(spans):
+def _unfold_segment(spans: list[Span]):
+    for span in spans:
         if span is not None:
             yield Value(
-                payload=KafkaPayload(key=None, value=orjson.dumps(span), headers=[]),
-                committable=committable if i == last else {},
+                KafkaPayload(key=None, value=orjson.dumps(span), headers=[]),
+                {},
                 timestamp=None,
             )