ref: bump sentry-arroyo to 2.38.5

pyproject.toml

@@ -81,7 +81,7 @@ dependencies = [
   "rfc3339-validator>=0.1.2",
   "rfc3986-validator>=0.1.1",
   # [end] jsonschema format validators
-  "sentry-arroyo>=2.38.1",
+  "sentry-arroyo>=2.38.5",
   "sentry-conventions>=0.3.0",
   "sentry-forked-email-reply-parser>=0.5.12.post1",
   "sentry-kafka-schemas>=2.1.27",

src/sentry/monitors/tasks/clock_pulse.py

@@ -8,7 +8,10 @@
 from arroyo import Partition
 from arroyo import Topic as ArroyoTopic
 from arroyo.backends.kafka import KafkaPayload
-from confluent_kafka.admin import AdminClient, PartitionMetadata
+from confluent_kafka.admin import (  # type: ignore[attr-defined]
+    AdminClient,
+    PartitionMetadata,
+)
 from django.conf import settings
 from sentry_kafka_schemas.codecs import Codec
 from sentry_kafka_schemas.schema_types.ingest_monitors_v1 import ClockPulse, IngestMonitorMessage

src/sentry/sentry_metrics/consumers/indexer/multiprocess.py

@@ -93,7 +93,7 @@ def submit(self, message: Message[KafkaPayload | FilteredPayload]) -> None:
             on_delivery=partial(
                 self.callback, committable=message.committable, timestamp=message.timestamp
             ),
-            headers=message.payload.headers,
+            headers=message.payload.headers,  # type: ignore[arg-type]
         )
 
     def callback(

src/sentry/sentry_metrics/consumers/indexer/routing_producer.py

@@ -118,7 +118,7 @@ def poll(self) -> None:
     def __delivery_callback(
         self,
         future: Future[str],
-        error: KafkaError,
+        error: KafkaError | None,
         message: ConfluentMessage,
     ) -> None:
         if error is not None:
@@ -142,14 +142,12 @@ def submit(self, message: Message[RoutingPayload]) -> None:
 
         future: Future[str] = Future()
         future.set_running_or_notify_cancel()
-        (
-            producer.produce(
-                topic=topic.name,
-                value=output_message.payload.value,
-                key=output_message.payload.key,
-                headers=output_message.payload.headers,
-                on_delivery=partial(self.__delivery_callback, future),
-            ),
+        producer.produce(
+            topic=topic.name,
+            value=output_message.payload.value,
+            key=output_message.payload.key,
+            headers=output_message.payload.headers,  # type: ignore[arg-type]
+            on_delivery=partial(self.__delivery_callback, future),
         )
         self.__queue.append((output_message.committable, future))
 

src/sentry/testutils/pytest/kafka.py

@@ -4,7 +4,8 @@
 from collections.abc import MutableMapping
 
 import pytest
-from confluent_kafka import Consumer, Producer
+from arroyo.processing import StreamProcessor
+from confluent_kafka import Producer
 from confluent_kafka.admin import AdminClient
 
 from sentry.testutils.pytest import xdist
@@ -73,7 +74,7 @@ def scope_consumers():
     be created once per test session).
 
     """
-    all_consumers: MutableMapping[str, Consumer | None] = {
+    all_consumers: MutableMapping[str, StreamProcessor | None] = {
         xdist.get_kafka_topic("ingest-events"): None,
         xdist.get_kafka_topic("outcomes"): None,
     }

src/sentry/utils/batching_kafka_consumer.py

@@ -15,7 +15,7 @@ def wait_for_topics(admin_client: AdminClient, topics: list[str], timeout: int =
     """
     for topic in topics:
         start = time.time()
-        last_error = None
+        last_error: str | KafkaError | None = None
 
         while True:
             if time.time() > start + timeout:
@@ -25,10 +25,14 @@ def wait_for_topics(admin_client: AdminClient, topics: list[str], timeout: int =
 
             result = admin_client.list_topics(topic=topic, timeout=timeout)
             topic_metadata = result.topics.get(topic)
-            if topic_metadata and topic_metadata.partitions and not topic_metadata.error:
+            if topic_metadata is None:
+                last_error = "Topic metadata not found"
+                time.sleep(0.1)
+                continue
+            if topic_metadata.partitions and not topic_metadata.error:
                 logger.debug("Topic '%s' is ready", topic)
                 break
-            elif topic_metadata.error in {
+            elif topic_metadata.error is not None and topic_metadata.error.code() in {
                 KafkaError.UNKNOWN_TOPIC_OR_PART,
                 KafkaError.LEADER_NOT_AVAILABLE,
             }:

tests/sentry/monitors/tasks/test_clock_pulse.py

@@ -3,7 +3,7 @@
 
 from arroyo import Partition, Topic
 from arroyo.backends.kafka import KafkaPayload
-from confluent_kafka.admin import PartitionMetadata
+from confluent_kafka.admin import PartitionMetadata  # type: ignore[attr-defined]
 from django.test import override_settings
 
 from sentry.monitors.tasks.clock_pulse import MONITOR_CODEC, clock_pulse