feat(consumer): add --retry-handle-destroyed flag

src/sentry/consumers/__init__.py

@@ -469,6 +469,7 @@ def get_stream_processor(
  add_global_tags: bool = False,
  profile_consumer_join: bool = False,
  enable_autocommit: bool = False,
+ retry_handle_destroyed: bool = False,
 ) -> StreamProcessor:
  from sentry.utils import kafka_config
 
@@ -527,6 +528,7 @@ def build_consumer_config(group_id: str):
  auto_offset_reset=auto_offset_reset,
  strict_offset_reset=strict_offset_reset,
  enable_auto_commit=enable_autocommit,
+ retry_handle_destroyed=retry_handle_destroyed,
  )
 
  if max_poll_interval_ms is not None:

src/sentry/runner/commands/run.py

@@ -556,6 +556,12 @@ def taskbroker_send_tasks(
  default=False,
  help="Enable Kafka autocommit mode with 1s commit interval. Offsets are stored via store_offsets and rdkafka commits them automatically.",
 )
+@click.option(
+ "--retry-handle-destroyed",
+ is_flag=True,
+ default=False,
+ help="Enable retrying on `KafkaError._DESTROY` during commit.",
+)
 @configuration
 def basic_consumer(
  consumer_name: str,