feat: Arroyo 2.12.0 (#4362)

Applying the basic retry policy with 3 retries everywhere now

requirements.txt

@@ -24,7 +24,7 @@ python-dateutil==2.8.2
 python-rapidjson==1.8
 pytz==2022.2.1
 redis==4.3.4
-sentry-arroyo==2.11.6
+sentry-arroyo==2.12.0
 sentry-kafka-schemas==0.1.10
 sentry-redis-tools==0.1.6
 sentry-relay==0.8.21

snuba/consumers/consumer_builder.py

@@ -16,8 +16,7 @@
 from arroyo.processing.strategies import ProcessingStrategyFactory
 from arroyo.types import Topic
 from arroyo.utils.profiler import ProcessingStrategyProfilerWrapperFactory
-from arroyo.utils.retries import BasicRetryPolicy, RetryPolicy
-from confluent_kafka import KafkaError, KafkaException, Producer
+from confluent_kafka import KafkaError, Producer
 from sentry_sdk.api import configure_scope
 
 from snuba.consumers.consumer import (
@@ -72,7 +71,6 @@ def __init__(
  metrics: MetricsBackend,
  slice_id: Optional[int],
  join_timeout: Optional[int],
- commit_retry_policy: Optional[RetryPolicy] = None,
  profile_path: Optional[str] = None,
  max_poll_interval_ms: Optional[int] = None,
  ) -> None:
@@ -143,21 +141,6 @@ def __init__(
  self.__profile_path = profile_path
  self.max_poll_interval_ms = max_poll_interval_ms
 
- if commit_retry_policy is None:
- commit_retry_policy = BasicRetryPolicy(
- 3,
- 1,
- lambda e: isinstance(e, KafkaException)
- and e.args[0].code()
- in (
- KafkaError.REQUEST_TIMED_OUT,
- KafkaError.NOT_COORDINATOR,
- KafkaError._WAIT_COORD,
- ),
- )
-
- self.__commit_retry_policy = commit_retry_policy
-
  def __build_consumer(
  self,
  strategy_factory: ProcessingStrategyFactory[KafkaPayload],
@@ -189,7 +172,6 @@ def log_general_error(e: KafkaError) -> None:
 
  consumer = KafkaConsumer(
  configuration,
- commit_retry_policy=self.__commit_retry_policy,
  )
 
  self.dlq_producer: Optional[KafkaProducer]